// Package cluster encapsulates all cluster functionality
package cluster

import (
	"context"
	"crypto/tls"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/gogo/protobuf/types"
	"github.com/pingcap/dbaas/central/config"
	"github.com/pingcap/dbaas/central/db"
	"github.com/pingcap/dbaas/central/db/dbutil"
	clusterDao "github.com/pingcap/dbaas/central/db/dbutil/cluster"
	"github.com/pingcap/dbaas/central/db/dbutil/useralerts"
	"github.com/pingcap/dbaas/central/db/entity"
	"github.com/pingcap/dbaas/central/db/entity/capacity"
	"github.com/pingcap/dbaas/central/db/tenant"
	"github.com/pingcap/dbaas/central/grpc/grpcli"
	infraapi "github.com/pingcap/dbaas/central/infra-api"
	"github.com/pingcap/dbaas/central/model"
	"github.com/pingcap/dbaas/central/model/cluster/clusterproxy/tidbcloud"
	tenantmodel "github.com/pingcap/dbaas/central/model/tenant"
	"github.com/pingcap/dbaas/central/model/util"
	"github.com/pingcap/dbaas/central/utils"
	"github.com/pingcap/dbaas/pkg/tidb/component"
	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/idl/pbgen/proto/base"
	"github.com/tidbcloud/idl/pbgen/proto/central"
	"github.com/tidbcloud/idl/pbgen/proto/core"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/log"
	mailv2 "github.com/tidbcloud/pkgv2/mail/v2"
	"github.com/tidbcloud/pkgv2/salesforce"
	"github.com/tidbcloud/pkgv2/secret/encrypt"
	pkgutil "github.com/tidbcloud/pkgv2/util"
	"go.opentelemetry.io/otel"
	"go.uber.org/multierr"
	"go.uber.org/zap"
	"gorm.io/gorm"
	"k8s.io/utils/pointer"
)

// https://cloud.google.com/compute/docs/troubleshooting/troubleshooting-vm-creation#resource_availability
const (
	ErrorInfraGcpZoneResourcePoolExhausted = "ZONE_RESOURCE_POOL_EXHAUSTED_WITH_DETAILS"
	ErrorInfraResourcesDepleted            = "ERR_INFRA_RESOURCES_DEPLETED"

	// get super read_only flag
	GetSuperReadonlyClause = "show global variables where variable_name = 'tidb_super_read_only';"
)

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

// Model is created with NewModel
type Model struct {
	log              *zap.Logger
	encryptor        encrypt.Encryptor
	mailFacade       mailv2.Facade
	salesforceClient *salesforce.Client
	tenantModel      tenantmodel.Model
	AESCrypt         *encrypt.AESCrypt
	ProxyTlsConfig   *tls.Config
	api              *infraapi.API
	dao              clusterDao.ClusterDaoFacade
	UserAlertService useralerts.UserAlertService
}

// NewModel returns a Model
func NewModel(log *zap.Logger,
	infraAPI *infraapi.API,
	encryptor encrypt.Encryptor,
	mailFacade mailv2.Facade,
	salesforceClient *salesforce.Client,
	tenantModel tenantmodel.Model,
	aesCrypt *encrypt.AESCrypt,
	proxyTlsConfig *tls.Config,
	dao clusterDao.ClusterDaoFacade,
) Model {
	return Model{
		log:              log,
		encryptor:        encryptor,
		mailFacade:       mailFacade,
		salesforceClient: salesforceClient,
		tenantModel:      tenantModel,
		AESCrypt:         aesCrypt,
		ProxyTlsConfig:   proxyTlsConfig,
		api:              infraAPI,
		dao:              dao,
		UserAlertService: &useralerts.DefaultUserAlertsServiceImpl{},
	}
}

type BaseContext interface {
	model() Model
}

// APIContext attaches a APIKey to a Context.
// This helps audit the APIKey
type APIContext struct {
	Model
	apiKeyEntity *entity.APIKey
	log          *zap.Logger
}

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

func (c APIContext) TenantID() uint64 {
	return c.apiKeyEntity.TenantID.Uint64()
}

func (c APIContext) model() Model {
	return c.Model
}

func (m Model) InfraAPI() *infraapi.API {
	return m.api
}

// API returns a APIContext
func (m Model) API(apiKeyEntity *entity.APIKey) APIContext {
	return APIContext{
		m,
		apiKeyEntity,
		m.log.With(
			zap.Uint64("tenantID", apiKeyEntity.TenantID.Uint64()),
			zap.String("apiKeyName", apiKeyEntity.Name),
			zap.String("module", "cluster")),
	}
}

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

func (c UserContext) TenantID() uint64 {
	return c.tenantAndUser.TenantID.Uint64()
}

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

func (m Model) Encryptor() encrypt.Encryptor {
	return m.encryptor
}

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

func (m Model) TenantLog(tenantID entity.ID) *zap.Logger {
	return m.log.With(zap.Uint64("tenantID", uint64(tenantID)), zap.String("module", "cluster"))
}

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

func (c UserContext) CheckOrgID(orgID entity.ID) error {
	if c.tenantAndUser.IsImpersonated {
		return nil
	}
	userTenantID := c.tenantAndUser.TenantID
	if orgID != userTenantID {
		return errcode.NewForbiddenErr(errors.New("org_id parameter is inconsistent with the user's org ID"))
	}
	return nil
}

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) model() Model {
	return c.Model
}

// EndpointStatus indicates if an endpoinDeleteJobt is ready
type EndpointStatus struct {
	Host  string
	IP    string
	Port  int32
	Ready bool
}

// Cluster is the model for a TiDB cluster.
type Cluster struct {
	ID              uint64
	Name            string
	Status          central.Cluster_Status
	SecondaryStatus central.Cluster_SecondaryStatus
	Progress        central.Cluster_CreateProgress
	Version         string
	Provider        entity.Cloud
	Region          string
	CreatedAt       time.Time
	CreatedBy       *entity.User
	Endpoints       Endpoints
	VPCCidrBlock    string
	Resources       []component.Resources
	Restored        bool
	// user will grant bucket permission to this RoleARN
	// and we will use this role to access user's bucket.
	RoleARN          string
	ImportID         *uint64
	LastErrors       []*central.LastError
	Replications     []*central.Replication
	ClusterType      central.Cluster_ClusterType
	WriteStatus      central.WriteStatus
	IsDevTier        bool
	XAccountDeployId uint64
}

// DeletedCluster is the model for a deleted TiDB cluster.
type DeletedCluster struct {
	ID               uint64
	Name             string
	Provider         string
	Region           string
	XAccountDeployId uint64
	DeleteTime       time.Time
	HasBackups       bool
	LatestBackupSize uint64
}

type TenantName int

// Get retrieves the cluster
func (c UserContext) Get(ctx context.Context, orgID, projectID, clusterID entity.ID) (*Cluster, error) {
	rdb := db.NewProvider().ReadDB(ctx)
	cluster, err := grpcli.GetCluster(ctx, orgID.Uint64(), projectID.Uint64(), clusterID.Uint64())
	if err != nil {
		return nil, err
	}

	restored := false
	restores, bizerr := dbutil.RestoresInProgress(rdb, clusterID)
	if bizerr != nil {
		return nil, bizerr
	}
	restored = len(restores) != 0

	cl := c.convertClusterV2(ctx, cluster, restored)
	return &cl, nil
}

// List all deleted clusters
func (c UserContext) ListDeletedClusters(ctx context.Context, provider entity.Cloud, projectID entity.ID) ([]*DeletedCluster, error) {
	var (
		clusters []entity.Cluster
		err      error
	)
	items := []*DeletedCluster{}

	err = db.Transaction(func(tx *gorm.DB) error {
		if string(provider) != "" {
			clusters, err = dbutil.ListDeletedClustersByProjectAndProvider(tx, provider, c.tenantAndUser.TenantID, projectID)
		} else {
			clusters, err = dbutil.ListDeletedClustersByProject(tx, c.tenantAndUser.TenantID, projectID)
		}
		return err
	})

	for _, cluster := range clusters {
		var region *entity.Region
		err = db.Transaction(func(tx *gorm.DB) error {
			region, err = dbutil.GetClusterRegion(ctx, tx, cluster)
			return err
		})
		if err != nil {
			return items, err
		}
		deletedCluster := c.convertDeletedCluster(cluster, region)
		if deletedCluster.HasBackups {
			items = append(items, deletedCluster)
		}
	}
	return items, nil
}

func (c UserContext) convertDeletedCluster(cluster entity.Cluster, region *entity.Region) *DeletedCluster {
	var (
		err     error
		backups []entity.Backup
	)
	deletedCluster := &DeletedCluster{
		ID:               cluster.ID.Uint64(),
		Name:             cluster.Name,
		Provider:         string(cluster.Provider.Name),
		Region:           region.Name,
		XAccountDeployId: cluster.CrossAccountDeploymentID.Uint64(),
		DeleteTime:       *cluster.DeletedTime,
		HasBackups:       false, // change later
		LatestBackupSize: 0,     // fill later
	}

	err = db.Transaction(func(tx *gorm.DB) error {
		backups, err = dbutil.ClusterBackups(tx, cluster.ID)
		return err
	})
	if err != nil || len(backups) == 0 {
		if err != nil {
			c.Log().Error("failed to list deleted backups", zap.Uint64("cluster id", cluster.ID.Uint64()), zap.Error(err))
		}
		deletedCluster.HasBackups = false
		return deletedCluster
	}
	deletedCluster.HasBackups = true

	dbBackupMap := make(map[string]bool, len(backups))
	for _, item := range backups {
		dbBackupMap[util.BackupCRDName(&item)] = true
	}

	backupCRs, err := c.api.ListBackups(context.TODO(), cluster.ProjectID.Uint64(), cluster.ID.Uint64())
	if err != nil {
		c.Log().Error("failed to list backups", zap.Uint64("clusterID", cluster.ID.Uint64()), zap.Error(err))
	} else {
		var size uint64
		for _, item := range backupCRs {
			if dbBackupMap[item.Name] && item.Status.BackupSize > 0 {
				size = uint64(item.Status.BackupSize) // FIXME: should compare `BackupTS` or other field to get the real LATEST backup?
			}
		}
		deletedCluster.LatestBackupSize = size
	}

	return deletedCluster
}

func (c UserContext) ListV2(ctx context.Context, projectID entity.ID, provider entity.Cloud, page, perPage uint32) ([]Cluster, int32, error) {
	filter := &core.ClusterFilter{
		TenantIds:  []uint64{c.tenantAndUser.TenantID.Uint64()},
		ProjectIds: []uint64{projectID.Uint64()},
		Deleted:    pointer.BoolPtr(false),
	}
	pager := &base.Pager{
		Page:     int64(page),
		PageSize: int64(perPage),
	}
	sorter := &base.Sorter{
		OrderBy: "created_at",
		Sort:    base.Sort_SORT_DESC,
	}
	if provider.String() != "" {
		filter.CloudProviderNames = []string{provider.String()}
	}

	listReq := &core.ListClustersReq{
		Filter:  filter,
		Pager:   pager,
		Sorter:  sorter,
		BaseReq: nil,
	}
	listResp, err := grpcli.ListClusters(ctx, listReq)
	if err != nil {
		return nil, 0, err
	}
	countResp, err := grpcli.CountClusters(ctx, &core.CountClustersReq{Filter: filter})
	if err != nil {
		return nil, 0, err
	}

	rdb := db.NewProvider().ReadDB(ctx)
	out := make([]Cluster, 0, len(listResp.Clusters))
	for _, clusterId := range listResp.OrderedClusterIds {
		restored := false
		restores, err := dbutil.RestoresInProgress(rdb, entity.ID(clusterId))
		if err != nil {
			return nil, 0, err
		}
		restored = len(restores) != 0
		out = append(out, c.convertClusterV2(ctx, listResp.Clusters[clusterId], restored))
	}
	return out, int32(countResp.Count), nil
}

func assembleReplicationList(ctx context.Context, clusterID entity.ID) ([]*central.Replication, central.Cluster_ClusterType, error) {
	var (
		err            error
		tmpReplication *central.Replication
		rList          []entity.Replication
		retList        []*central.Replication
		clusterType    central.Cluster_ClusterType
	)
	clusterType = central.Cluster_COMMON
	err = db.Transaction(func(tx *gorm.DB) error {
		err = entity.NewReplicationQuerySet(tx).
			PrimaryClusterIDEq(clusterID).
			PreloadPrimaryCluster().
			PreloadSecondaryCluster().
			PreloadDetachUser().
			PreloadSecondaryUser().
			All(&rList)
		if err != nil {
			return err
		}
		for _, r := range rList {
			if r.ReplicationStatus != entity.ReplicationStatusDetached {
				clusterType = central.Cluster_PRIMARY
			}
			tmpReplication, err = assembleReplication(ctx, clusterID, r)
			if err != nil {
				return err
			}
			retList = append(retList, tmpReplication)
		}
		err = entity.NewReplicationQuerySet(tx).
			SecondaryClusterIDEq(clusterID).
			PreloadPrimaryCluster().
			PreloadSecondaryCluster().
			PreloadDetachUser().
			PreloadSecondaryUser().
			All(&rList)
		if err != nil {
			return err
		}
		for _, r := range rList {
			if r.ReplicationStatus != entity.ReplicationStatusDetached {
				clusterType = central.Cluster_SECONDARY
			}
			tmpReplication, err = assembleReplication(ctx, clusterID, r)
			if err != nil {
				return err
			}
			retList = append(retList, tmpReplication)
		}
		return nil
	})
	return retList, clusterType, err
}

func assembleReplication(ctx context.Context, clusterID entity.ID, r entity.Replication) (*central.Replication, error) {
	var (
		err      error
		ret      *central.Replication
		tracking entity.ReplicationTracking
	)
	ret = &central.Replication{
		PrimaryClusterName:   r.PrimaryCluster.Name,
		PrimaryClusterId:     r.PrimaryCluster.ID.Uint64(),
		Status:               string(r.ReplicationStatus),
		ReplicationId:        r.ID.Uint64(),
		ReplicationLag:       "unknow",
		PrimaryWriteStatus:   central.WriteStatus_WRITABLE,
		SecondaryWriteStatus: central.WriteStatus_WRITABLE,
	}
	if r.PrimaryCluster.ReadOnly&entity.ReadOnlyStatus != 0 {
		ret.PrimaryWriteStatus = central.WriteStatus_READONLY
	}
	if r.SecondaryCluster != nil {
		ret.SecondaryClusterName = r.SecondaryCluster.Name
		ret.SecondaryClusterId = r.SecondaryCluster.ID.Uint64()
		if r.SecondaryCluster.ReadOnly&entity.ReadOnlyStatus != 0 {
			ret.SecondaryWriteStatus = central.WriteStatus_READONLY
		}
	}
	if r.SecondaryUser != nil {
		ret.Creator = r.SecondaryUser.Email
	}
	if r.DetachUser != nil {
		ret.DetachUser = r.DetachUser.Email
	}
	if r.LagTimeWhenDetach != nil {
		ret.ReplicationLag = strconv.Itoa(int(*r.LagTimeWhenDetach))
	}
	err = db.Transaction(func(tx *gorm.DB) error {
		return entity.NewReplicationTrackingQuerySet(tx).
			ReplicationIDEq(r.ID).
			StatusEq(entity.ReplicationStatusDetaching).
			One(&tracking)
	})
	if err != nil && err != gorm.ErrRecordNotFound {
		log.Debug(ctx, "failed to get tracking of replication", zap.Uint64("replication id", r.ID.Uint64()))
	} else if err == nil {
		ret.DetachedTime, err = types.TimestampProto(tracking.CreatedAt)
		if err != nil {
			log.Debug(ctx, "failed to parse detached time of tracking", zap.Uint64("replication id", r.ID.Uint64()))
		}
	}
	ret.CreationTime = utils.MustTimestampProto(r.CreatedAt)
	if r.ReplicationStatus == entity.ReplicationStatusCreating {
		ret.CreateProgress = getReplicationProgress(clusterID, &r)
	}

	return ret, err
}

func (c UserContext) addLastError(cluster *core.Cluster, model *Cluster) error {
	cr, err := c.api.GetCluster(context.TODO(), cluster.ProjectId, cluster.Id)
	if err != nil {
		return errors.Annotate(err, "GetCluster")
	}

	status, err := infraapi.ConvertClusterStatus(cluster.InfraProviderType, cr)
	if err != nil {
		return errors.Annotate(err, "ConvertClusterStatus")
	}

	lastErrors := status.GardenerStatus.ShootLastErrors

	// Currently, we only check ERR_INFRA_RESOURCES_DEPLETED errors
	// https://github.com/gardener/gardener/blob/v1.19.1/pkg/apis/core/v1beta1/types_common.go#L36
	// GCP: check lastError.Description
	for _, lastError := range lastErrors {
		var infraResourcesDepleted bool
		for _, code := range lastError.Codes {
			if code == ErrorInfraResourcesDepleted {
				infraResourcesDepleted = true
				break
			}
		}
		if !infraResourcesDepleted && strings.Contains(lastError.Description, ErrorInfraGcpZoneResourcePoolExhausted) {
			infraResourcesDepleted = true
		}

		if infraResourcesDepleted {
			infraLastError := central.LastError{
				Codes: []central.InfraErrorCode{
					central.InfraErrorCode_INFRA_RESOURCES_DEPLETED,
				},
			}
			model.LastErrors = append(model.LastErrors, &infraLastError)
			break
		}
	}

	return nil
}

func (c UserContext) convertClusterV2(ctx context.Context, cluster *core.Cluster, restored bool) Cluster {
	rdb := db.NewProvider().ReadDB(ctx)
	logger := log.WithContextS(ctx)
	model := Cluster{
		ID:               cluster.Id,
		Name:             cluster.Name,
		Status:           central.Cluster_Status(cluster.UserStatus),
		SecondaryStatus:  central.Cluster_SecondaryStatus(cluster.UserSubStatus),
		Progress:         central.Cluster_COMPLETED,
		Version:          cluster.Version,
		Provider:         entity.Cloud(cluster.CloudProviderName),
		Region:           cluster.RegionName,
		CreatedAt:        time.Unix(cluster.CreatedAt, 0),
		CreatedBy:        nil,
		Endpoints:        Endpoints{},
		VPCCidrBlock:     "",
		Resources:        util.CoreClusterComponents2componentResources(cluster.Components),
		Restored:         restored,
		RoleARN:          "",
		ImportID:         cluster.ImportId,
		LastErrors:       nil,
		Replications:     nil,
		ClusterType:      0,
		WriteStatus:      0,
		IsDevTier:        cluster.IsDevTier,
		XAccountDeployId: cluster.XaccountDeployId,
	}

	if cluster.CreatedBy != 0 {
		creator, err := dbutil.GetUserByID(rdb, entity.ID(cluster.TenantId), entity.ID(cluster.CreatedBy))
		if err != nil {
			logger.Warnf("failed to get creator of cluster %d: %s", cluster.Id, err)
		}
		model.CreatedBy = &creator
	}

	switch cluster.Status {
	case core.ClusterStatus_CLUSTER_STATUS_INITIALIZING:
		model.Progress = central.Cluster_INITIALIZING
	case core.ClusterStatus_CLUSTER_STATUS_PREPARING:
		model.Progress = central.Cluster_PREPARING
	case core.ClusterStatus_CLUSTER_STATUS_LAUNCHING:
		model.Progress = central.Cluster_LAUNCHING
	}

	var ready bool
	if cluster.UserStatus != core.UserClusterStatus_USER_CLUSTER_STATUS_CREATING {
		ready = true
		model.Endpoints.PrometheusReady = true
		model.Endpoints.PdReady = true
	}

	for _, each := range cluster.Connections {
		if each.Type == core.ClusterConnectionType_CLUSTER_CONNECTION_TYPE_PUBLIC {
			model.Endpoints.MySQL = EndpointStatus{
				Host:  each.Host,
				Port:  each.Port,
				Ready: ready,
			}
		} else if each.Type == core.ClusterConnectionType_CLUSTER_CONNECTION_TYPE_PRIVATE {
			model.Endpoints.PrivateMySQL = EndpointStatus{
				Host:  each.Host,
				Port:  each.Port,
				Ready: ready,
			}
		}
	}

	if cluster.UserStatus == core.UserClusterStatus_USER_CLUSTER_STATUS_CREATING {
		if err := c.addLastError(cluster, &model); err != nil {
			c.Log().Error("fail to add last error to TiDB cluster", zap.Error(err))
		}
	}

	// add replications to cluster entity
	var err error
	model.Replications, model.ClusterType, err = assembleReplicationList(ctx, entity.ID(cluster.Id))
	if err != nil {
		c.Log().Error("failed to attach replication to cluster", zap.Uint64("clusterID", cluster.Id))
		return model
	}

	// for the secondary cluster, the status need be modify via the replication status
	// when the replication status is BACKUPING/RESTOREING/REPLICATION_PREPAREING the cluster status is creating
	if model.Progress == central.Cluster_COMPLETED {
		for _, r := range model.Replications {
			if r.Status == string(entity.ReplicationStatusCreating) {
				model.Progress = r.CreateProgress
				if model.ID == r.SecondaryClusterId {
					model.Status = central.Cluster_Status(core.UserClusterStatus_USER_CLUSTER_STATUS_CREATING)
				}
			}
		}
	}

	network, err := c.api.GetNetwork(context.TODO(), cluster.InfraProviderType, cluster.ProjectId, cluster.RegionName)
	if err != nil {
		c.Log().Error("failed to get network", zap.Error(err))
		return model
	}

	networkStatus, err := infraapi.ConvertNetworkStatus(network)
	if err != nil {
		c.Log().Error("failed to convert network status", zap.Error(err))
		return model
	}

	model.VPCCidrBlock = networkStatus.VPCCidrBlock
	switch model.Provider {
	case entity.Aws:
		model.RoleARN = networkStatus.ImportRoleARN
	case entity.Gcp:
		model.RoleARN = networkStatus.ServiceAccountEmail
	}

	model.WriteStatus = central.WriteStatus_WRITABLE
	if cluster.ReadOnly {
		model.WriteStatus = central.WriteStatus_READONLY
	}

	return model
}

// getReplciationProgress return the cluster progress when in create replication progress
// BACKUPING/RESTOREING/REPLICATION_PREPAREING is only show in secondary cluster create progress
func getReplicationProgress(clusterID entity.ID, r *entity.Replication) central.Cluster_CreateProgress {
	if clusterID != r.SecondaryClusterID {
		return central.Cluster_COMPLETED
	}
	if r.BackupID == 0 {
		return central.Cluster_BACKUPING
	} else if r.CDCTaskID == 0 {
		return central.Cluster_RESTORING
	} else {
		return central.Cluster_REPLICATION_PREPARING
	}
}

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

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

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

func (c BgContext) TenantID() uint64 {
	return c.tenantAndUser.TenantID.Uint64()
}

func (c BgContext) model() Model {
	return c.Model
}

// Delete soft deletes a cluster in the database. Teardown will remove the running cluster.
func (c UserContext) Delete(ctx context.Context, orgID, projectID, clusterID entity.ID) error {
	var replication entity.Replication
	// just check cluster exist
	return db.Transaction(func(tx *gorm.DB) error {
		model.LockTenant(tx, orgID)

		cluster, err := dbutil.GetClusterByID(tx, orgID, projectID, clusterID)
		if err != nil {
			if errors.IsNotFound(err) {
				return nil
			}
			return err
		}
		// the primary and secondary clsuter can't be delete before detached
		if cluster.ReplicationID != nil && entity.EmptyReplicationID != *cluster.ReplicationID {
			err = entity.NewReplicationQuerySet(tx).
				IDEq(entity.ID(*cluster.ReplicationID)).
				ReplicationStatusNe(entity.ReplicationStatusDetached).
				One(&replication)
			if err != nil {
				return err
			}
			// if this cluster is in create replication
			// DBaaS need to detach it first then delete the cluster
			if cluster.ID == replication.PrimaryClusterID {
				return errors.New("This Cluster is a Primary Cluster of Cross Region Replication. Detach the Secondary Cluster before deleting the Primary Cluster")
			} else if replication.ReplicationStatus == entity.ReplicationStatusReplicating && cluster.ID == replication.SecondaryClusterID {
				return errors.New("This Cluster is a Secondary Cluster of Cross Region Replication. Detach it before deleting it")
			}

			_, err = c.DetachReplication(orgID, clusterID, replication.ID, central.DetachReplicationReq_None)
			if err != nil {
				return errcode.NewInternalErr(errors.Annotate(err, "fail to delete cluster"))
			}
		}
		if err := dbutil.DeleteCluster(ctx, tx, clusterID); err != nil {
			return err
		}
		return dbutil.AddAccessLogNow(tx, c.tenantAndUser.User.TenantID, c.tenantAndUser.Email, fmt.Sprintf("delete cluster: %s<%d>", cluster.Name, clusterID))
	})
}

func (c UserContext) validateChangeClusterName(clusterID entity.ID, name string) error {
	return db.Transaction(func(tx *gorm.DB) error {
		count, err := entity.NewClusterQuerySet(tx).IDNe(clusterID).NameEq(name).TenantIDEq(c.tenantAndUser.TenantID).DeletedTimeIsNull().Count()
		if err != nil {
			return errcode.NewInternalErr(errors.New("failed to query cluster count by name"))
		}
		if count > 0 {
			return errcode.NewInvalidInputErr(errors.New(dbutil.ErrUniqueClusterName))
		}
		return nil
	})
}

func (c UserContext) ChangeClusterName(orgID entity.ID, projectID entity.ID, clusterID entity.ID, clusterName string) error {
	err := c.validateChangeClusterName(clusterID, clusterName)
	if err != nil {
		c.log.Error("validateClusterName", zap.Uint64("clusterID", clusterID.Uint64()), zap.Error(err))
		return err
	}

	err = db.Transaction(func(tx *gorm.DB) error {
		model.LockTenant(tx, orgID)

		cluster, err := dbutil.GetClusterByID(tx, orgID, projectID, clusterID)
		if err != nil {
			return errors.Annotate(err, "failed to get cluster by ID")
		}
		if cluster.Name == clusterName {
			return nil
		}

		cluster.Name = clusterName
		if err := cluster.Update(tx, entity.ClusterDBSchema.Name); err != nil {
			return errors.Annotate(err, "failed to update cluster name")
		}
		return dbutil.AddAccessLogNow(tx, c.tenantAndUser.User.TenantID, c.tenantAndUser.Email, fmt.Sprintf("change cluster: %d, name: %s", clusterID, clusterName))
	})
	if err != nil {
		c.log.Error("failed to change cluster name", zap.Uint64("clusterID", clusterID.Uint64()), zap.Error(err))
	}
	return nil
}

func changeClusterPasswordKey(email string, clusterID entity.ID) string {
	return fmt.Sprintf("%d-%s", clusterID.Uint64(), email)
}

func (c UserContext) SendPasswordVerifyCode(ctx context.Context, orgID, projectID, clusterID entity.ID) error {
	var (
		cluster entity.Cluster
		err     error
	)
	err = db.Transaction(func(tx *gorm.DB) error {
		cluster, err = dbutil.GetClusterByID(tx, orgID, projectID, clusterID)
		return err
	})
	if err != nil {
		return err
	}

	randCode := pkgutil.RandString(6)
	email := c.tenantAndUser.User.Email
	clusterName := cluster.Name

	if err := model.InsertCode(ctx, changeClusterPasswordKey(email, clusterID), randCode); err != nil {
		return err
	}
	vars := map[string]string{
		"code":    randCode,
		"cluster": clusterName,
	}
	err = c.mailFacade.SendMailWithTemplate(ctx, []string{email}, config.EmailSubjectPasswordVerificationCode, config.TNResetPasswordVerifyCode, vars, nil)
	if err != nil {
		log.Error(ctx, "failed to send password verify code email", zap.String("email", email), zap.Time("time", time.Now().UTC()), zap.Error(err))
		return err
	}
	log.Info(ctx, "send password verify code email",
		zap.String("email", email),
		zap.Time("time", time.Now().UTC()),
	)
	return nil
}

// ChangePassword changes the user's password
func (c UserContext) ChangePassword(ctx context.Context, orgID entity.ID, projectID entity.ID, clusterID entity.ID, newPassword, code string) error {
	var cluster entity.Cluster

	err := db.Transaction(func(tx *gorm.DB) error {
		var err error
		cluster, err = dbutil.GetClusterByID(tx, orgID, projectID, clusterID)
		if err != nil {
			c.log.Error("failed to get cluster by ID", zap.Uint64("clusterID", clusterID.Uint64()), zap.Error(err))
			return errcode.NewInternalErr(errors.Errorf("failed to get cluster by ID %d", clusterID))
		}
		if cluster.Status != entity.ClusterStatusNormal {
			return errcode.NewInvalidInputErr(errors.New("cluster status is not normal"))
		}

		return dbutil.AddAccessLogNow(tx, c.tenantAndUser.User.TenantID, c.tenantAndUser.Email, fmt.Sprintf("change password for cluster: %d", clusterID))
	})
	if err != nil {
		return err
	}

	if !model.ValidateCode(ctx, changeClusterPasswordKey(c.tenantAndUser.User.Email, clusterID), code) {
		return errcode.NewInvalidInputErr(errors.New("Verification code is not correct"))
	}

	cr, _, err := FindClusterStatus(ctx, c.api, clusterID, &orgID)
	if err != nil {
		return errors.AddStack(err)
	}
	adminPwd, err := util.DecryptCloudAdminPassword(c.encryptor, cluster, cr.Status)
	if err != nil {
		return errors.AddStack(err)
	}

	err = BgContext(c).changeClusterPassword(context.TODO(), cluster, adminPwd, newPassword)
	if err != nil {
		c.log.Error("failed to change root password of cluster", zap.Uint64("clusterID", clusterID.Uint64()), zap.Error(err))
		return errcode.NewInternalErr(errors.New("failed to change root password of cluster"))
	}
	return nil
}

// Endpoints is the collection of endpoints for the Cluster.
type Endpoints struct {
	MySQL           EndpointStatus
	PrometheusReady bool
	PdReady         bool
	PrivateMySQL    EndpointStatus
}

func FindClusterStatus(ctx context.Context, api *infraapi.API, clusterID entity.ID, orgID *entity.ID) (*cloudv1.Cluster, entity.Cluster, error) {
	var cluster entity.Cluster

	err := db.Transaction(func(tx *gorm.DB) error {
		var err error
		cluster, err = dbutil.GetCluster(tx, clusterID)
		if err != nil {
			return err
		}

		if orgID != nil && cluster.TenantID != *orgID {
			return errcode.NewForbiddenErr(fmt.Errorf("cluster %d does not belong to tenant %d", clusterID, orgID))
		}

		return nil
	})

	if err != nil {
		return nil, cluster, err
	}

	cr, err := api.GetCluster(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
	if err != nil {
		return nil, cluster, err
	}
	return cr, cluster, nil
}

func FindClusterMetricsBaseURL(ctx context.Context, api *infraapi.API, clusterID entity.ID, orgID *entity.ID) (string, error) {
	cr, _, err := FindClusterStatus(ctx, api, clusterID, orgID)
	if err != nil {
		return "", errors.AddStack(err)
	}

	if cr.Status.Metrics == nil {
		return "", errors.New("no metrics status")
	}

	return cr.Status.Metrics.BaseURL, nil
}

func FindClusterDashboardBaseURL(ctx context.Context, encryptor encrypt.Encryptor, api *infraapi.API, clusterID entity.ID, orgID *entity.ID) (string, error) {
	cr, cluster, err := FindClusterStatus(ctx, api, clusterID, orgID)
	if err != nil {
		return "", errors.Annotatef(err, "did not find cluster status for cluster %d", uint(clusterID))
	}
	if cr.Status.Proxy == nil {
		return "", errors.Errorf("no proxy status for cluster %d", uint(clusterID))
	}

	params := ""
	adminPwd, err := util.DecryptCloudAdminPassword(encryptor, cluster, cr.Status)
	if err != nil {
		return "", errors.Annotate(err, "could not decrypt password")
	}
	if len(adminPwd) > 0 {
		params = "?password=" + base64.StdEncoding.EncodeToString([]byte(string(adminPwd)))
	} else {
		log.Info(ctx, "no cloud admin password for cluster", zap.Uint64("clusterID", uint64(clusterID)))
	}

	return cr.Status.Proxy.Endpoint + params, nil
}

func FindClusterProxyBaseURL(ctx context.Context, api *infraapi.API, clusterID entity.ID, orgID *entity.ID) (string, error) {
	cr, _, err := FindClusterStatus(ctx, api, clusterID, orgID)
	if err != nil {
		return "", errors.AddStack(err)
	}
	if cr.Status.Proxy == nil {
		return "", errors.New("no proxy status")
	}
	return cr.Status.Proxy.Endpoint, nil
}

// getComponentName trims the tc name prefix
// "{tcname}-tidb-0" -> "tidb-0"
func getComponentNodeName(podName string) string {
	ar := strings.Split(podName, "-")
	if len(ar) > 2 {
		ar = ar[len(ar)-2:]
	}
	return strings.Join(ar, "-")
}

func tikvNodeStatus(ctx context.Context, s cloudv1.MemberStatus) central.TiKVNode_Status {
	switch s {
	case cloudv1.MemberStatusNormal:
		return central.TiKVNode_NORMAL
	case cloudv1.MemberStatusCreating:
		return central.TiKVNode_CREATING
	case cloudv1.MemberStatusUnavailable:
		return central.TiKVNode_UNAVAILABLE
	case cloudv1.MemberStatusTerminating:
		return central.TiKVNode_TERMINATING
	case cloudv1.MemberStatusLeaving:
		return central.TiKVNode_LEAVING
	default:
		log.Error(ctx, "unknown member status for tikv", zap.String("status", string(s)))
		return central.TiKVNode_NORMAL
	}
}

func tiflashNodeStatus(ctx context.Context, s cloudv1.MemberStatus) central.TiFlashNode_Status {
	switch s {
	case cloudv1.MemberStatusNormal:
		return central.TiFlashNode_NORMAL
	case cloudv1.MemberStatusCreating:
		return central.TiFlashNode_CREATING
	case cloudv1.MemberStatusUnavailable:
		return central.TiFlashNode_UNAVAILABLE
	case cloudv1.MemberStatusTerminating:
		return central.TiFlashNode_TERMINATING
	case cloudv1.MemberStatusLeaving:
		return central.TiFlashNode_LEAVING
	default:
		log.Error(ctx, "unknown member status for tiflash", zap.String("status", string(s)))
		return central.TiFlashNode_NORMAL
	}
}

func tidbNodeStatus(ctx context.Context, s cloudv1.MemberStatus) central.TiDBNode_Status {
	switch s {
	case cloudv1.MemberStatusNormal:
		return central.TiDBNode_NORMAL
	case cloudv1.MemberStatusCreating:
		return central.TiDBNode_CREATING
	case cloudv1.MemberStatusUnavailable:
		return central.TiDBNode_UNAVAILABLE
	case cloudv1.MemberStatusTerminating:
		return central.TiDBNode_TERMINATING
	default:
		log.Error(ctx, "unknown member status for tidb", zap.String("status", string(s)))
		return central.TiDBNode_NORMAL
	}
}

func (c UserContext) ListTiKVNodesV1(ctx context.Context, orgID, projectID, clusterID entity.ID) (nodes []*central.TiKVNode, err error) {
	var cluster entity.Cluster
	err = db.Transaction(func(tx *gorm.DB) error {
		var err error
		cluster, err = dbutil.GetClusterByID(tx, orgID, projectID, clusterID)
		return err
	})

	if err != nil {
		return nil, err
	}

	cr, err := c.api.GetCluster(ctx, cluster.ProjectID.Uint64(), clusterID.Uint64())
	if err != nil {
		return nil, err
	}

	if cr.Status.DB == nil {
		return nil, nil
	}

	com := cr.Status.DB.TiKV
	if com == nil {
		return nil, nil
	}

	// don't use the storageSize of cr, use component.StorageSizeGi instead.
	// it's not correct for local-storage, due to the some space are reserved by filesystem.
	storageSizeGi := uint32(0)
	for _, each := range cluster.ComponentsV2 {
		if each.ComponentType == entity.TiKV {
			storageSizeGi = each.StorageSizeGi
		}
	}
	mUnit := int64(MemoryUnit)
	for _, m := range com.Members {
		node := &central.TiKVNode{
			Name:      getComponentNodeName(m.Name),
			Status:    tikvNodeStatus(ctx, m.Status),
			Type:      central.TiKVNode_STORAGE,
			StorageGi: storageSizeGi,
			Compute: &central.Compute{
				Cpu:      float32(m.Resources.CPU.Value()),
				MemoryMi: uint32(m.Resources.Memory.Value() / mUnit / mUnit),
			},
			Az: m.Labels["cloud.pingcap.com/zone"],
		}

		nodes = append(nodes, node)
	}

	return nodes, nil
}

func (c UserContext) ListTiFlashNodesV1(ctx context.Context, orgID, projectID, clusterID entity.ID) (nodes []*central.TiFlashNode, err error) {
	var cluster entity.Cluster
	err = db.Transaction(func(tx *gorm.DB) error {
		var err error
		cluster, err = dbutil.GetClusterByID(tx, orgID, projectID, clusterID)
		return err
	})

	if err != nil {
		return nil, err
	}

	cr, err := c.api.GetCluster(ctx, cluster.ProjectID.Uint64(), clusterID.Uint64())
	if err != nil {
		return nil, err
	}

	if cr.Status.DB == nil {
		return nil, nil
	}

	com := cr.Status.DB.TiFlash
	if com == nil {
		return nil, nil
	}

	// don't use the storageSize of cr, use component.StorageSizeGi instead.
	// it's not correct for local-storage, due to the some space are reserved by filesystem.
	storageSizeGi := uint32(0)
	for _, each := range cluster.ComponentsV2 {
		if each.ComponentType == entity.TiFlash {
			storageSizeGi = each.StorageSizeGi
		}
	}
	mUnit := int64(MemoryUnit)
	for _, m := range com.Members {
		node := &central.TiFlashNode{
			Name:      getComponentNodeName(m.Name),
			Status:    tiflashNodeStatus(ctx, m.Status),
			Type:      central.TiFlashNode_STORAGE,
			StorageGi: storageSizeGi,
			Compute: &central.Compute{
				Cpu:      float32(m.Resources.CPU.Value()),
				MemoryMi: uint32(m.Resources.Memory.Value() / mUnit / mUnit),
			},
			Az: m.Labels["cloud.pingcap.com/zone"],
		}

		nodes = append(nodes, node)
	}

	return nodes, nil
}

func (c UserContext) ListTiDBNodesV1(ctx context.Context, orgID, projectID, clusterID entity.ID) (nodes []*central.TiDBNode, err error) {
	var cluster entity.Cluster
	err = db.Transaction(func(tx *gorm.DB) error {
		var err error
		cluster, err = dbutil.GetClusterByID(tx, orgID, projectID, clusterID)
		return err
	})

	if err != nil {
		return nil, err
	}

	cr, err := c.api.GetCluster(ctx, cluster.ProjectID.Uint64(), clusterID.Uint64())
	if err != nil {
		return nil, err
	}

	if cr.Status.DB == nil {
		return nil, nil
	}

	com := cr.Status.DB.TiDB
	if com == nil {
		return nil, nil
	}

	mUnit := int64(MemoryUnit)
	for _, m := range com.Members {
		node := &central.TiDBNode{
			Name:   getComponentNodeName(m.Name),
			Status: tidbNodeStatus(ctx, m.Status),
			Type:   central.TiDBNode_COMPUTING,
			Compute: &central.Compute{
				Cpu:      float32(m.Resources.CPU.Value()),
				MemoryMi: uint32(m.Resources.Memory.Value() / mUnit / mUnit),
			},
			Az: m.Labels["cloud.pingcap.com/zone"],
		}

		nodes = append(nodes, node)
	}

	return nodes, nil
}

func (c BgContext) GetClusterVersionV1(ctx context.Context, projectID uint64, clusterID uint64) (version string, err error) {
	cluster, err := c.api.GetCluster(ctx, projectID, clusterID)
	if err != nil {
		return "", err
	}
	version, err = infraapi.GetClusterVersion(cluster)
	if err != nil {
		c.Log().Error("get cluster version", zap.Error(err))
	}
	return version, nil
}

const (
	MemoryUnit uint32 = 1024
)

func GetClusterComponentsV2(cluster entity.Cluster) ([]component.Resources, error) {
	var componentResources []component.Resources

	err := db.Transaction(func(tx *gorm.DB) error {
		var clusterComponents []*entity.ClusterComponent
		if err := tx.Where("cluster_id=?", cluster.ID).
			Preload("LabelProfile").
			Preload("InternalProfile").Preload("InternalProfile.InfraCompute").Preload("InternalProfile.InfraStorage").
			Find(&clusterComponents).Error; err != nil {
			return err
		}
		for _, each := range clusterComponents {
			compute := each.InternalProfile.InfraCompute
			storage := each.InternalProfile.InfraStorage
			machineType := capacity.MachineType(compute.InstanceType)
			componentResources = append(componentResources, component.Resources{
				ComponentName: component.Name(each.ComponentType),
				Image:         nil,
				Capacity: capacity.Capacity{
					MachineType: &machineType,
					Replicas:    each.Replicas,
					Compute: &capacity.Compute{
						MemoryMi: compute.MemorySizeGi * 1024,
						CPU:      float32(compute.CPUNum),
					},
					Disk: &capacity.Disk{
						SizeGi: each.StorageSizeGi,
						Type:   capacity.DiskType(storage.InstanceType),
					},
					Node: &capacity.Node{
						NodeProfileID: each.LabelProfile.ID.Uint64(),
						Name:          each.LabelProfile.Name,
					},
				},
			})
		}
		return nil
	})

	return componentResources, err
}

func (c UserContext) DetachReplication(orgID, clusterID, rID entity.ID, detachType central.DetachReplicationReq_DetachType) (*central.DetachReplicationResp, error) {
	var (
		err error
		r   entity.Replication
	)
	err = db.Transaction(func(tx *gorm.DB) error {
		err = entity.NewReplicationQuerySet(tx).
			IDEq(rID).
			PreloadPrimaryCluster().
			PreloadSecondaryCluster().
			One(&r)
		if err != nil {
			return err
		}

		// if cluster is not primary or secondary cluster of replication
		// return input error
		if r.SecondaryClusterID != clusterID && r.PrimaryClusterID != clusterID {
			return errcode.NewInvalidInputErr(errors.Errorf("cluster %d is not primary or secondary cluster of replication %d", clusterID, rID))
		}

		// planned detach need to set primary as read-only and wait for the upstream and downstream replication is algin
		// TODO:wait for read-only function is ready
		if detachType == central.DetachReplicationReq_Planned {
			r.DetachType = entity.DetachTypePlanned
		} else if detachType == central.DetachReplicationReq_Force {
			r.DetachType = entity.DetachTypeForce
		} else if detachType == central.DetachReplicationReq_None {
			r.DetachType = entity.DetachTypeNone
		} else {
			return errcode.NewInvalidInputErr(errors.Errorf("unknow replication type %d", detachType))
		}

		// update replication, primary cluster and secondary cluster status as detach other gc worker
		r.ReplicationStatus = entity.ReplicationStatusDetaching
		r.DetachUserID = c.tenantAndUser.User.ID
		r.PrimaryCluster.ReplicationID = nil
		if err = r.PrimaryCluster.Update(tx, entity.ClusterDBSchema.ReplicationID); err != nil {
			return errors.Annotate(err, "failed to update cluster replication ID")
		}
		if r.SecondaryClusterID != 0 {
			// secondary cluster stop receive binlog
			// use delete private svc to do this
			// and the replicaiton secondary cluster could also delete in the `creating` status
			// so DBaaS should make judgement weather the secondary cluster is exist here
			r.SecondaryCluster.ReplicationID = nil
			if err = r.SecondaryCluster.Update(tx, entity.ClusterDBSchema.ReplicationID); err != nil {
				return errors.Annotate(err, "failed to update cluster replication ID")
			}
		}
		return dbutil.UpdateReplication(tx, &r)
	})
	if err != nil {
		return nil, err
	}

	return &central.DetachReplicationResp{
		ReplicationId: rID.Uint64(),
		ClusterId:     clusterID.Uint64(),
	}, err
}

func (c BgContext) CleanClusters(ctx context.Context) error {
	var clusters []entity.Cluster
	err := db.Transaction(func(tx *gorm.DB) error {
		return entity.NewClusterQuerySet(tx).StatusNotIn(
			entity.DeletedOrClearedClusterStatus...).TenantIDEq(c.tenantAndUser.TenantID).FreeEq(false).All(&clusters)
	})
	if err != nil {
		c.Log().Error("trial worker: failed to list active clusters of user", zap.Error(err))
		return errors.Annotate(err, "failed to list active clusters of user")
	}

	var errs error
	for _, cluster := range clusters {
		err := db.Transaction(func(tx *gorm.DB) error {
			// Only make final backups for poc clusters in normal status
			if cluster.Status == entity.ClusterStatusNormal {
				return dbutil.DeleteClusterAndMakeFinalBackup(ctx, tx, cluster.ID)
			}
			return dbutil.DeleteCluster(ctx, tx, cluster.ID)
		})
		if err != nil {
			c.Log().Error("failed to delete cluster", zap.String("clusterName", cluster.Name), zap.Uint64("clusterID", uint64(cluster.ID)), zap.Error(err))
			errs = multierr.Append(errs, errors.Annotatef(err, "failed to delete cluster %d", cluster.ID))
			continue
		}
		c.Log().Info("delete cluster", zap.String("clusterName", cluster.Name), zap.Uint64("clusterID", uint64(cluster.ID)))
	}
	return errs
}

func (c BgContext) CleanFreeClusters(ctx context.Context) error {
	var clusters []entity.Cluster
	err := db.Transaction(func(tx *gorm.DB) error {
		return entity.NewClusterQuerySet(tx).StatusNe(entity.ClusterStatusCleared).TenantIDEq(c.tenantAndUser.TenantID).FreeEq(true).All(&clusters)
	})

	if err != nil {
		return err
	}
	var errs error
	for _, cluster := range clusters {
		if cluster.DeletedTime == nil {
			err := db.Transaction(func(tx *gorm.DB) error {
				return dbutil.DeleteCluster(ctx, tx, cluster.ID)
			})

			if err != nil {
				c.Log().Error("failed to delete cluster", zap.String("clusterName", cluster.Name), zap.Uint64("clusterID", uint64(cluster.ID)), zap.Error(err))
				errs = multierr.Append(errs, errors.Annotatef(err, "failed to delete cluster %d", cluster.ID))
				continue
			}
			c.Log().Info("delete cluster", zap.String("clusterName", cluster.Name), zap.Uint64("clusterID", uint64(cluster.ID)))
		}
	}

	if errs != nil {
		return errs
	}
	allClusterIDs := make([]entity.ID, 0, len(clusters))
	for _, each := range clusters {
		allClusterIDs = append(allClusterIDs, each.ID)
	}

	// TODO(chenfei): backups should be deleted automatically when the cluster be deleted
	// or provide an option to determine whether put it in the recycle bin or delete all resource
	// deleted cluster may have some backups, so delete all uncleared cluster's backups
	err = db.Transaction(func(tx *gorm.DB) error {
		return dbutil.DeleteBackupsByClusterIDs(tx, allClusterIDs)
	})

	if err != nil {
		c.Log().Error("failed to delete cluster backups", zap.Error(err))
		return err
	}
	return nil
}

/*
 * SyncClearCDCInfo
 * after force detach, when upstream cluster up, need clear cdc info, or not will affect late replications
 */
func (c BgContext) SyncClearCDCInfo(ctx context.Context) error {
	var (
		err      error
		tenantID entity.ID
		clusters []entity.Cluster
	)
	tenantID = c.tenantAndUser.TenantID
	clusters, err = c.dao.FindClusters(dbutil.ClusterFilter{
		TenantID: &tenantID,
		Deleted:  pointer.BoolPtr(false),
		Cleared:  pointer.BoolPtr(false),
		Creating: pointer.BoolPtr(false),
	})
	if err != nil {
		return errors.Annotatef(err, "failed to fetch clusters records")
	}
	for _, cluster := range clusters {
		if cluster.ReplicationID != nil && entity.EmptyReplicationID == *cluster.ReplicationID {
			c.log.Info(fmt.Sprintf("begin clear cdc info of cluster %d", cluster.ID))
			if err = c.ClearCDCInfo(ctx, &cluster, tidbcloud.BuildClusterProxy); err != nil {
				c.log.Error(fmt.Sprintf("failed to clear cdc info of cluster %d", cluster.ID), zap.Error(err))
				return err
			}
			cluster.ReplicationID = nil
			err = c.dao.UpdateCluster(cluster, []entity.ClusterDBSchemaField{entity.ClusterDBSchema.ReplicationID}, true)
			if err != nil {
				c.log.Error(fmt.Sprintf("failed to update cluster %d replicationID nil", cluster.ID), zap.Error(err))
				return err
			}
			c.log.Info(fmt.Sprintf("success clear cdc info of cluster %d", cluster.ID))
		}
	}
	return nil
}

func (c BgContext) ClearCDCInfo(ctx context.Context, cluster *entity.Cluster, getProxy util.GetProxyFunc) error {
	var (
		err   error
		cr    *cloudv1.Cluster
		proxy clusterproxy.ClusterProxy
	)

	cr, err = c.api.GetCluster(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
	if err != nil {
		return errors.Annotatef(err, "failed to get cluster")
	}

	if cr == nil {
		return nil
	}

	proxy, err = getProxy(cluster.ID.Uint64(), cr.Status, config.DefaultClusterProxyTimeout)
	if err != nil {
		return errors.Annotatef(err, "failed to get proxy")
	}

	err = proxy.ClearCDCInfo(ctx)
	if err != nil {
		return errors.Annotatef(err, "failed to clear cdc info")
	}

	return nil
}

// TODO: (yinliang) combine it with replication
type ValueModel struct {
	Variable []string `json:"Value"`
}

func (c BgContext) GetReadOnly(ctx context.Context, cluster *entity.Cluster, getProxy util.GetProxyFunc) (bool, error) {
	var (
		err             error
		data, decodeRet []byte
		adminPwd        encrypt.Secret
		cr              *cloudv1.Cluster
		proxy           clusterproxy.ClusterProxy
		superReadOnly   ValueModel
	)

	cr, err = c.api.GetCluster(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
	if err != nil {
		return false, errors.Annotatef(err, "failed to get cluster")
	}

	if cr == nil {
		return false, nil
	}

	proxy, err = getProxy(cluster.ID.Uint64(), cr.Status, config.DefaultClusterProxyTimeout)
	if err != nil {
		return false, errors.Annotatef(err, "failed to get proxy")
	}

	adminPwd, err = util.DecryptCloudAdminPassword(c.encryptor, *cluster, cr.Status)
	if err != nil {
		return false, errors.Annotatef(err, "failed to decrypt admin password")
	}

	data, err = proxy.QuerySQL(ctx, adminPwd, GetSuperReadonlyClause)
	if err != nil {
		return false, errors.Annotatef(err, "failed to query super read only")
	}

	err = json.Unmarshal(data, &superReadOnly)
	if err != nil {
		return false, errors.Annotatef(err, "failed to unmarshal super read only")
	}

	if len(superReadOnly.Variable) == 0 {
		return false, nil
	}

	decodeRet, err = base64.StdEncoding.DecodeString(superReadOnly.Variable[0])
	if err != nil {
		return false, errors.Annotatef(err, "failed to decrypted super read only")
	}

	if string(decodeRet) == "OFF" || string(decodeRet) == "off" {
		return false, nil
	}
	return true, nil
}

func (c BgContext) SyncReadonly(ctx context.Context) error {
	var (
		err      error
		readonly bool
		tenantID entity.ID
		clusters []entity.Cluster
	)
	tenantID = c.tenantAndUser.TenantID
	clusters, err = c.dao.FindClusters(dbutil.ClusterFilter{
		TenantID: &tenantID,
		Deleted:  pointer.BoolPtr(false),
		Cleared:  pointer.BoolPtr(false),
		Creating: pointer.BoolPtr(false),
	})
	if err != nil {
		return errors.Annotatef(err, "failed to fetch clusters records")
	}
	for _, cluster := range clusters {
		if cluster.IsDevTier() {
			continue
		}
		// if force set readonly, make cluster readonly and set ForceReadOnlyStatus zero
		if cluster.ReadOnly&entity.ForceReadOnlyStatus != 0 {
			err = UserContext(c).SetReadOnly(ctx, cluster.ID, cluster.ProjectID, "ON")
			if err != nil {
				c.log.Error("failed to set read only status", zap.Error(err))
				continue
			}
			cluster.ReadOnly &= (^entity.ForceReadOnlyStatus)
			return c.dao.UpdateCluster(cluster, []entity.ClusterDBSchemaField{entity.ClusterDBSchema.ReadOnly}, true)
		}

		// else just sync cluster read only status from cluster to db
		readonly, err = c.GetReadOnly(ctx, &cluster, tidbcloud.BuildClusterProxy)
		if err != nil {
			// don't block cluster sync for the proxy problems
			c.log.Error("failed to get read only status", zap.Error(err))
			continue
		}
		oldReadonly := cluster.ReadOnly
		if readonly {
			cluster.ReadOnly |= entity.ReadOnlyStatus
		} else {
			cluster.ReadOnly &= (^entity.ReadOnlyStatus)
		}
		// don't update db if readonly status is not changed
		if cluster.ReadOnly == oldReadonly {
			continue
		}
		err = c.dao.UpdateCluster(cluster, []entity.ClusterDBSchemaField{entity.ClusterDBSchema.ReadOnly}, false)
		if err != nil {
			c.log.Error("failed to update cluster read only status", zap.Error(err))
			continue
		}
	}
	return nil
}

// MakeWritable change the cluster's read_only status
func (c UserContext) SetReadOnly(ctx context.Context, clusterID entity.ID, projectID entity.ID, trigger string) error {
	var (
		err     error
		cluster entity.Cluster
	)
	cluster, err = c.dao.GetCluster(clusterID)
	if err != nil {
		return err
	}

	if cluster.ProjectID != projectID {
		return errcode.NewForbiddenErr(errors.New("cluster_id parameter is inconsistent with the user's project ID"))
	}
	return util.ExecuteSqlCommon(ctx, &cluster, fmt.Sprintf(config.SetSuperReadonlyClause, trigger), c.api, c.encryptor, tidbcloud.BuildClusterProxy, config.DefaultClusterProxyTimeout)
}

func (c UserContext) CheckIfUserIsAuditAdmin() error {
	user := c.tenantAndUser
	if user.Role == entity.UserRoleAuditAdmin {
		return nil
	}
	return errcode.NewForbiddenErr(errors.New("only audit admin can perform this operation"))
}

func QueryClusterWithSQL(ctx context.Context, api *infraapi.API, encryptor encrypt.Encryptor, cluster *entity.Cluster, getProxy util.GetProxyFunc, sql string) ([]byte, error) {
	var (
		err      error
		adminPwd encrypt.Secret
		cr       *cloudv1.Cluster
		proxy    clusterproxy.ClusterProxy
	)

	cr, err = api.GetCluster(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
	if err != nil {
		return nil, errors.Annotatef(err, "failed to get cluster")
	}

	proxy, err = getProxy(cluster.ID.Uint64(), cr.Status, config.DefaultClusterProxyTimeout)
	if err != nil {
		return nil, errors.Annotatef(err, "failed to get proxy")
	}

	adminPwd, err = util.DecryptCloudAdminPassword(encryptor, *cluster, cr.Status)
	if err != nil {
		return nil, errors.Annotatef(err, "failed to decrypt admin password")
	}

	return proxy.QuerySQL(ctx, adminPwd, sql)
}

func (c BgContext) QueryGlobalVariable(ctx context.Context, cluster *entity.Cluster, getProxy util.GetProxyFunc, sql string) ([]string, error) {
	var variable ValueModel

	data, err := QueryClusterWithSQL(ctx, c.api, c.encryptor, cluster, getProxy, sql)
	if err != nil {
		return nil, err
	}

	err = json.Unmarshal(data, &variable)
	if err != nil {
		return nil, errors.Annotatef(err, "failed to unmarshal variables")
	}

	return variable.Variable, nil
}

func (c BgContext) GetAuditLogStatus(ctx context.Context, cluster *entity.Cluster, getProxy util.GetProxyFunc) (bool, error) {
	variables, err := c.QueryGlobalVariable(ctx, cluster, getProxy, config.GetAuditLogStatusClause)
	if err != nil {
		return false, err
	}

	if len(variables) == 0 {
		return false, nil
	}

	decodeRet, err := base64.StdEncoding.DecodeString(variables[0])
	if err != nil {
		return false, errors.Annotatef(err, "failed to decrypt variables")
	}

	return strings.ToUpper(string(decodeRet)) == config.AuditRedactLogStatusOFF, nil
}