package replication

import (
	"context"
	"crypto/tls"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"strconv"
	"time"

	"github.com/pingcap/dbaas/central/config"
	"github.com/pingcap/dbaas/central/db"
	"github.com/pingcap/dbaas/central/db/dbutil"
	cdcdao "github.com/pingcap/dbaas/central/db/dbutil/cdc"
	clusterdao "github.com/pingcap/dbaas/central/db/dbutil/cluster"
	dao "github.com/pingcap/dbaas/central/db/dbutil/replication"
	"github.com/pingcap/dbaas/central/db/entity"
	"github.com/pingcap/dbaas/central/db/tenant"
	"github.com/pingcap/dbaas/central/grpc/grpcli"
	api "github.com/pingcap/dbaas/central/infra-api"
	"github.com/pingcap/dbaas/central/metrics"
	cdcModel "github.com/pingcap/dbaas/central/model/cdc"
	clusterModel "github.com/pingcap/dbaas/central/model/cluster"
	"github.com/pingcap/dbaas/central/model/cluster/clusterproxy/tidbcloud"
	tenantmodel "github.com/pingcap/dbaas/central/model/tenant"
	"github.com/pingcap/dbaas/central/model/util"
	modelutil "github.com/pingcap/dbaas/central/model/util"
	"github.com/pingcap/dbaas/central/service/cvt"
	"github.com/pingcap/dbaas/pkg/constants"
	"github.com/pingcap/errcode"
	perrors "github.com/pingcap/errors"
	"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"
	"github.com/tidbcloud/pkgv2/secret/encrypt"
	"go.uber.org/multierr"
	"go.uber.org/zap"
	"gorm.io/gorm"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
)

// Model is created with NewModel
type Model struct {
	log         *zap.Logger
	encryptor   encrypt.Encryptor
	transport   *HttpTransport
	tlsConfig   *tls.Config
	api         api.InfraFacade
	tenantModel tenantmodel.Model
	dao         dao.ReplicationDaoFacade
	clusterdao  clusterdao.ClusterDaoFacade
	cdcdao      cdcdao.CDCDaoFacade
}

// NewModel returns a Model
func NewModel(
	log *zap.Logger,
	encryptor encrypt.Encryptor,
	tlsConfig *tls.Config,
	api api.InfraFacade,
	tenantModel tenantmodel.Model,
	dao dao.ReplicationDaoFacade,
	clusterdao clusterdao.ClusterDaoFacade,
	cdcdao cdcdao.CDCDaoFacade) Model {
	return Model{
		log:       log,
		encryptor: encryptor,
		transport: &HttpTransport{
			HttpClient: &http.Client{
				Timeout: 10 * time.Second,
				Transport: &http.Transport{
					TLSClientConfig: tlsConfig,
				},
			},
			Authenticator: &Auth0Authenticator{
				Auth0Domain:       config.Global().ProxyConfig.Auth0Domain,
				Auth0ClientID:     config.Global().ProxyConfig.Auth0ClientID,
				Auth0ClientSecret: config.Global().ProxyConfig.Auth0ClientSecret,
				Auth0Audience:     config.Global().ProxyConfig.Auth0Audience,
			},
		},
		tlsConfig:   tlsConfig,
		api:         api,
		tenantModel: tenantModel,
		dao:         dao,
		clusterdao:  clusterdao,
		cdcdao:      cdcdao,
	}
}

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

// User returns a UserContext
func (m Model) User(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", "replication"),
	)}
}

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

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
}

// 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{User: systemUser, TenantID: tenantID}, m.log.With(zap.Uint64("tenantID", tenantID.Uint64()), zap.Uint64("userID", uint64(systemUser.ID)), zap.String("module", "replication"))}, nil
}

func (c *BgContext) Run(ctx context.Context) error {
	creatingStatus := entity.ReplicationStatusCreating
	replications, err := c.dao.FindReplication(dao.ReplicationFilter{
		TenantID: c.tenantAndUser.TenantID.Uint64(),
		Status:   &creatingStatus,
	})
	if err != nil {
		return err
	}
	var errs error
	for _, r := range replications {
		err = c.DoCreate(ctx, &r)
		if err != nil {
			c.log.Debug("failed to create replication", zap.Uint64("replication ID", r.ID.Uint64()), zap.Error(err))
			errs = multierr.Append(errs, err)
		}
	}
	return errs
}

func (c BgContext) SetGCLifeTime(
	ctx context.Context,
	r *entity.Replication,
	duration string,
	getProxy util.GetProxyFunc,
) error {
	var (
		err      error
		adminPwd encrypt.Secret
		cr       *cloudv1.Cluster
		proxy    clusterproxy.ClusterProxy
	)

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

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

	adminPwd, err = modelutil.DecryptCloudAdminPassword(c.encryptor, *r.PrimaryCluster, cr.Status)
	if err != nil {
		return perrors.Annotatef(err, "failed to decrypt admin password")
	}

	return proxy.ExecuteSQL(ctx, adminPwd, fmt.Sprintf(config.SetGCLifeTimeClause, duration))
}

type VariableModel struct {
	Variable []string `json:"VARIABLE_VALUE"`
}

type ValueModel struct {
	Value []string `json:"Value"`
}

func (c BgContext) GetGCLifeTime(ctx context.Context, r *entity.Replication, getProxy util.GetProxyFunc) (string, error) {
	var (
		err             error
		data, decodeRet []byte
		adminPwd        encrypt.Secret
		cr              *cloudv1.Cluster
		proxy           clusterproxy.ClusterProxy
		gclifetime      VariableModel
	)

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

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

	adminPwd, err = modelutil.DecryptCloudAdminPassword(c.encryptor, *r.PrimaryCluster, cr.Status)
	if err != nil {
		return "", perrors.Annotatef(err, "failed to decrypt admin password")
	}

	data, err = proxy.QuerySQL(ctx, adminPwd, config.GetGCLifeTimeClause)
	if err != nil {
		return "", perrors.Annotatef(err, "failed to query gc life time")
	}

	err = json.Unmarshal(data, &gclifetime)
	if err != nil {
		return "", perrors.Annotatef(err, "failed to unmarshal gc life time")
	}

	decodeRet, err = base64.StdEncoding.DecodeString(gclifetime.Variable[0])
	if err != nil {
		return "", perrors.Annotatef(err, "failed to decrypted pre gc life time")
	}

	return string(decodeRet), nil
}

func (c *BgContext) DoCreate(ctx context.Context, r *entity.Replication) error {
	var (
		err           error
		preGcLifeTime string
	)

	// get previous gc life time before replication
	if r.PreGclifetime == "" {
		preGcLifeTime, err = c.GetGCLifeTime(ctx, r, tidbcloud.BuildClusterProxy)
		if err != nil {
			return err
		}

		r.PreGclifetime = preGcLifeTime
		err = c.dao.UpdateReplication(*r, []entity.ReplicationDBSchemaField{entity.ReplicationDBSchema.PreGclifetime})
		if err != nil {
			return err
		}
	}

	// change gc life time save point to 72h
	err = c.SetGCLifeTime(ctx, r, config.ReplicationGcLifeTime, tidbcloud.BuildClusterProxy)
	if err != nil {
		return err
	}
	if r.BackupID == 0 {
		return c.OnBackup(ctx, r)
	} else if r.BackupID != 0 && r.SecondaryClusterID == 0 {
		return c.OnRestore(ctx, r)
	} else if r.SecondaryClusterID != 0 && r.VPCPeeringID == nil {
		err = c.OnCreateVpcPeering(ctx, r)
		if err != nil {
			return err
		}
	} else if r.VPCPeeringID != nil && r.CDCTaskID == 0 {
		err = c.OnCheckVpcPeering(ctx, r)
		if err != nil {
			return err
		}
		err = c.OnCheckCluster(r.SecondaryCluster)
		if err != nil {
			return err
		}
		if err = modelutil.ExecuteSqlCommon(
			ctx,
			r.SecondaryCluster,
			fmt.Sprintf(config.SetRestrictReadonlyClause, "ON"),
			c.api,
			c.encryptor,
			tidbcloud.BuildClusterProxy, config.DefaultClusterProxyTimeout,
		); err != nil {
			return perrors.Annotatef(err, "failed to set restrict readonly")
		}
		return c.OnCreateCDC(ctx, r)
	} else if r.CDCTaskID != 0 {
		err = c.OnCheckCDC(r)
		if err != nil {
			return err
		}
		err = c.SetGCLifeTime(ctx, r, r.PreGclifetime, tidbcloud.BuildClusterProxy)
		if err != nil {
			return err
		}
		r.ReplicationStatus = entity.ReplicationStatusReplicating
		return c.dao.UpdateReplication(*r, []entity.ReplicationDBSchemaField{entity.ReplicationDBSchema.ReplicationStatus})
	}
	return nil
}

func (c *BgContext) OnBackup(ctx context.Context, r *entity.Replication) error {
	var (
		err               error
		backup            entity.Backup
		pCluster          entity.Cluster
		newBackupSchedule *entity.BackupSchedule
	)
	newBackupSchedule = &entity.BackupSchedule{
		Type:        entity.BackupType_Manual,
		ClusterID:   r.PrimaryClusterID,
		UserID:      c.tenantAndUser.User.ID,
		Method:      entity.BackupMethod_Full,
		Description: "backup for cross region replication",
		Name:        fmt.Sprintf("replication%d", r.ID),
	}

	return db.Transaction(func(tx *gorm.DB) error {
		err = entity.NewClusterQuerySet(dbutil.PreloadCluster(tx)).
			IDEq(r.PrimaryClusterID).
			One(&pCluster)
		if err != nil {
			return perrors.Annotate(err, "failed to get primary cluster")
		}
		if err = newBackupSchedule.Create(tx); err != nil {
			return perrors.Annotate(err, "failed to create backup shedulebackup")
		}
		err = dbutil.CreateBackup(ctx, tx, *newBackupSchedule, pCluster)
		if err != nil {
			return perrors.Annotate(err, "failed to create backup")
		}
		err = entity.NewBackupQuerySet(tx).BackupScheduleIDEq(newBackupSchedule.ID).One(&backup)
		if err != nil {
			return perrors.Annotate(err, "failed to get backup")
		}
		r.BackupID = backup.ID
		return c.dao.UpdateReplication(*r, []entity.ReplicationDBSchemaField{entity.ReplicationDBSchema.BackupID})
	})
}

func (c *BgContext) OnCreateCDC(ctx context.Context, r *entity.Replication) error {
	var (
		err            error
		adminPwd       encrypt.Secret
		connector      *central.ConnectorID
		startTimeStamp uint64
		endpoint       string
	)

	endpoint = fmt.Sprintf("db-tidb-peer.tidb%d", r.SecondaryClusterID.Uint64())
	startTimeStamp, err = strconv.ParseUint(r.Backup.StartTimeStamp, 10, 64)
	if err != nil {
		return err
	}

	cr, err := c.api.GetCluster(ctx, r.PrimaryCluster.ProjectID.Uint64(), r.PrimaryClusterID.Uint64())
	if err != nil {
		return perrors.AddStack(err)
	}
	adminPwd, err = modelutil.DecryptCloudAdminPassword(c.encryptor, *r.SecondaryCluster, cr.Status)
	if err != nil {
		return perrors.Annotatef(err, "failed to decrypt admin password")
	}

	// ticdc create in the downstream cluster
	// so we have to use secondary cluster ID here
	connector, err = cdcModel.CreateCdcTaskImpl(
		r.SecondaryClusterID,
		central.ConnectorType_TIDB,
		fmt.Sprintf("%s:%d", endpoint, 4000),
		"",
		entity.CloudAdminUser,
		string(adminPwd),
		startTimeStamp,
		c.clusterdao,
		c.encryptor,
		c.log,
		r.IgnoreReplication)
	if err != nil {
		return err
	}
	r.CDCTaskID = entity.ID(connector.ConnectorId)
	return c.dao.UpdateReplication(*r, []entity.ReplicationDBSchemaField{entity.ReplicationDBSchema.CDCTaskID})
}

func (c *BgContext) OnCheckCDC(r *entity.Replication) error {
	var (
		err  error
		task *entity.CDCTask
	)
	if r.CDCTaskID == 0 {
		return perrors.Errorf("cdc task is not start yet")
	}
	task, err = c.cdcdao.GetCdcTaskByID(r.CDCTaskID, r.SecondaryCluster.ID, r.SecondaryCluster.ProjectID)
	if err != nil {
		return err
	}
	if task.Status != entity.CDCStatusNormal {
		return perrors.Errorf("cdc is not ready, status %s", task.Status)
	}
	return nil
}

// if the backup is done, create the secondary cluster by restore the primary cluster
func (c *BgContext) OnRestore(ctx context.Context, r *entity.Replication) error {
	var (
		err      error
		user     entity.User
		pCluster entity.Cluster
	)
	if err != nil {
		c.log.Error("failed to get organization", zap.Error(err))
		return errcode.NewInternalErr(errors.New("failed to get organization"))
	}
	return db.Transaction(func(tx *gorm.DB) error {
		err = db.Transaction(func(tx *gorm.DB) error {
			return entity.NewClusterQuerySet(dbutil.PreloadCluster(tx)).
				IDEq(r.PrimaryClusterID).
				One(&pCluster)
		})
		if err != nil {
			return err
		}
		err = entity.NewUserQuerySet(tx).IDEq(r.SecondaryUserID).One(&user)
		if err != nil {
			return err
		}
		secondaryPassword, err := encrypt.DecryptBase64(c.encryptor, r.SecondaryPassword)
		if err != nil {
			return err
		}
		replicationId := r.ID.Uint64()
		coreReq := &core.RestoreClusterReq{
			BackupId:           r.BackupID.Uint64(),
			OperatorUserId:     user.ID.Uint64(),
			ReplicationId:      &replicationId,
			RegionName:         &r.SecondaryRegion,
			ClusterName:        r.SecondaryName,
			ClusterDescription: "_",
			ClusterPassword:    string(secondaryPassword),
			Components:         cvt.ClusterComponentState2Core(r.SecondaryClusterComponents.Components),
			Port:               r.SecondaryClusterComponents.Port,
			BaseReq:            nil,
		}
		coreResp, err := grpcli.RestoreCluster(ctx, coreReq)
		if err != nil {
			return err
		}
		r.SecondaryClusterID = entity.ID(coreResp.ClusterId)
		return c.dao.UpdateReplication(*r, []entity.ReplicationDBSchemaField{entity.ReplicationDBSchema.SecondaryClusterID})
	})
}

func (c *BgContext) OnCheckCluster(cluster *entity.Cluster) error {
	if cluster.Status != entity.ClusterStatusNormal {
		return perrors.Errorf("wait for cluster ready")
	}
	return nil
}

func (c *BgContext) OnCheckVpcPeering(ctx context.Context, r *entity.Replication) error {
	var (
		err      error
		pcluster entity.Cluster
		pregion  *entity.Region
	)

	pcluster, err = c.clusterdao.GetCluster(r.PrimaryClusterID)
	if err != nil {
		return err
	}
	pregion, err = c.clusterdao.GetClusterRegion(ctx, pcluster)
	if err != nil {
		return err
	}

	// if replication in same region, no vpc peering created
	if pregion.Name == r.SecondaryRegion {
		return nil
	}

	peering, err := grpcli.GetNetworkPeer(ctx, pcluster.Provider.Name.String(), *r.VPCPeeringID)
	if err != nil {
		return err
	}
	if peering.Status != core.NetworkPeer_STATUS_ACTIVE {
		return perrors.Errorf("vpcpeering is not active, status %s", peering.Status)
	}
	return nil
}

func (c *BgContext) OnCreateVpcPeering(ctx context.Context, r *entity.Replication) error {
	// create cross vpc peering
	var (
		err                     error
		pcluster                entity.Cluster
		scluster                entity.Cluster
		tmpcluster              entity.Cluster
		snetwork                *cloudv1.Network
		createNetworkPeerParams *clusterModel.CreateNetworkPeerParams
		snetworkStatus          *cloudv1.GardenerNetworkStatus
		pregion                 *entity.Region
		sregion                 *entity.Region
		tmpregion               *entity.Region
	)

	pcluster, err = c.clusterdao.GetCluster(r.PrimaryClusterID)
	if err != nil {
		return err
	}

	scluster, err = c.clusterdao.GetCluster(r.SecondaryClusterID)
	if err != nil {
		return err
	}

	pregion, err = c.clusterdao.GetClusterRegion(ctx, pcluster)
	if err != nil {
		return err
	}

	sregion, err = c.clusterdao.GetClusterRegion(ctx, scluster)
	if err != nil {
		return err
	}

	// if same region, no vpc peering created
	if pregion.ID == sregion.ID {
		zeroUint64 := uint64(0)
		r.VPCPeeringID = &zeroUint64
		return c.dao.UpdateReplication(*r, []entity.ReplicationDBSchemaField{entity.ReplicationDBSchema.VPCPeeringID})
	} else if pregion.ID > sregion.ID {
		tmpregion = pregion
		pregion = sregion
		sregion = tmpregion
		tmpcluster = pcluster
		pcluster = scluster
		scluster = tmpcluster
	}

	if pcluster.Provider.Name == entity.Aws {
		snetwork, err = c.api.GetNetwork(context.TODO(), api.ProviderTypeAWS, scluster.ProjectID.Uint64(), sregion.Name)
		if err != nil {
			c.log.Error("query network failed", zap.Uint64("project", scluster.ProjectID.Uint64()), zap.Error(err))
			return err
		}

		snetworkStatus, err = cloudv1.GardenerNetworkStatusFromNetwork(snetwork)
		if err != nil {
			return err
		}
		if snetworkStatus.AWS.VpcID == nil {
			return perrors.Errorf("network is not ready")
		}

		createNetworkPeerParams = &clusterModel.CreateNetworkPeerParams{
			ProjectId:     pcluster.ProjectID.Uint64(),
			CloudProvider: pcluster.Provider.Name.String(),
			Region:        pregion.Name,
			PeerRegion:    snetwork.Spec.Region,
			PeerVpcCidr:   snetwork.Spec.CIDR,
			PeerNetwork:   *snetworkStatus.AWS.VpcID,
			Type:          core.NetworkPeer_TYPE_DBAAS,
			Aws: clusterModel.CreateNetworkPeerParams4Aws{
				PeerAccountId: *snetworkStatus.AWS.AccountID,
			},
		}
	} else {
		snetwork, err = c.api.GetNetwork(context.TODO(), api.ProviderTypeGCP, scluster.ProjectID.Uint64(), sregion.Name)
		if err != nil {
			c.log.Error("query network failed", zap.Uint64("project", scluster.ProjectID.Uint64()), zap.Error(err))
			return err
		}

		snetworkStatus, err = cloudv1.GardenerNetworkStatusFromNetwork(snetwork)
		if err != nil {
			return err
		}
		if snetworkStatus.GCP.ProjectID == nil {
			return perrors.Errorf("network is not ready")
		}
		createNetworkPeerParams = &clusterModel.CreateNetworkPeerParams{
			ProjectId:     pcluster.ProjectID.Uint64(),
			CloudProvider: pcluster.Provider.Name.String(),
			Region:        pregion.Name,
			PeerRegion:    snetwork.Spec.Region,
			PeerVpcCidr:   snetwork.Spec.CIDR,
			PeerNetwork:   *snetworkStatus.GCP.NetworkName,
			Type:          core.NetworkPeer_TYPE_DBAAS,
			Gcp: clusterModel.CreateNetworkPeerParams4Gcp{
				PeerProjectId: *snetworkStatus.GCP.ProjectID,
			},
		}
	}

	peeringId, err := clusterModel.CreateNetworkPeerImp(ctx, createNetworkPeerParams, c.tenantAndUser)
	if err != nil {
		return err
	}
	r.VPCPeeringID = &peeringId
	err = c.dao.UpdateReplication(*r, []entity.ReplicationDBSchemaField{entity.ReplicationDBSchema.VPCPeeringID})
	if err != nil {
		return err
	}
	return nil
}

func (c *BgContext) Clean(ctx context.Context) error {
	var errs error
	detachingStatus := entity.ReplicationStatusDetaching
	replications, err := c.dao.FindReplication(dao.ReplicationFilter{
		TenantID: c.tenantAndUser.TenantID.Uint64(),
		Status:   &detachingStatus,
	})
	if err != nil {
		return err
	}
	for _, r := range replications {
		err = c.DoDetach(ctx, r)
		if err != nil {
			c.log.Debug("failed to clean replication", zap.Uint64("replication ID", r.ID.Uint64()), zap.Error(err))
			errs = multierr.Append(errs, err)
		}
	}
	return errs
}

// gc all resources as below
func (c *BgContext) DoDetach(ctx context.Context, r entity.Replication) error {
	var (
		err           error
		track         entity.ReplicationTracking
		checkPointTSO uint64
	)

	// set primary cluster super read only
	if r.DetachType == entity.DetachTypePlanned {
		if err = modelutil.ExecuteSqlCommon(
			ctx,
			r.PrimaryCluster,
			fmt.Sprintf(config.SetSuperReadonlyClause, "ON"),
			c.api,
			c.encryptor,
			tidbcloud.BuildClusterProxy,
			config.DefaultClusterProxyTimeout,
		); err != nil {
			return perrors.Annotatef(err, "failed to set super readonly")
		}
	} else if r.DetachType == entity.DetachTypeForce {
		r.PrimaryCluster.ReadOnly |= entity.ForceReadOnlyStatus
		emptyReplicationID := entity.EmptyReplicationID
		r.PrimaryCluster.ReplicationID = &emptyReplicationID
		err = c.clusterdao.UpdateCluster(*r.PrimaryCluster, []entity.ClusterDBSchemaField{entity.ClusterDBSchema.ReadOnly, entity.ClusterDBSchema.ReplicationID}, true)
		if err != nil {
			return err
		}
	}

	if r.DetachType == entity.DetachTypePlanned {
		// first we need to record detach tso in the database
		if r.DetachTso == 0 {
			err = db.Transaction(func(tx *gorm.DB) error {
				r.DetachTso, err = c.GetClusterTSO(ctx, r.PrimaryCluster, tidbcloud.BuildClusterProxy)
				if err != nil {
					return perrors.Annotatef(err, "failed to get cluster tso")
				}
				return c.dao.UpdateReplication(r, []entity.ReplicationDBSchemaField{entity.ReplicationDBSchema.DetachTso})
			})
			if err != nil {
				return err
			}
		}

		// then get the tso of cdc and judge whether we have gap between primary and secondary
		if r.DetachCheckpointTso == 0 {
			return db.Transaction(func(tx *gorm.DB) error {
				checkPointTSO, err = c.GetCheckPointTSO(ctx, r.SecondaryCluster.ProjectID, r.SecondaryClusterID, r.CDCTaskID, tidbcloud.BuildClusterProxy)
				if err != nil {
					return perrors.Annotatef(err, "failed to get check point tso")
				}
				if checkPointTSO < r.DetachTso {
					return perrors.Errorf("secondary cluster is not catch up with the primary one detachTSO %d check point TSO %d", r.DetachTso, checkPointTSO)
				}
				r.DetachCheckpointTso = checkPointTSO
				return c.dao.UpdateReplication(r, []entity.ReplicationDBSchemaField{entity.ReplicationDBSchema.DetachCheckpointTso})
			})
		}
	}

	if err = c.OnDeleteCDC(ctx, r); err != nil {
		return perrors.Annotatef(err, "failed to delete cdc")
	}

	if r.DetachType == entity.DetachTypePlanned {
		if err = c.ensureTiCDCDeleted(ctx, r.SecondaryCluster); err != nil {
			return perrors.Annotatef(err, "wait cdc resource deleted")
		}
	}

	if r.DetachType != entity.DetachTypeNone {
		if err = modelutil.ExecuteSqlCommon(
			ctx,
			r.SecondaryCluster,
			fmt.Sprintf(config.SetRestrictReadonlyClause, "OFF"),
			c.api,
			c.encryptor,
			tidbcloud.BuildClusterProxy,
			config.DefaultClusterProxyTimeout,
		); err != nil {
			return perrors.Annotatef(err, "failed to set restrict readonly")
		}

		if err = modelutil.ExecuteSqlCommon(
			ctx,
			r.SecondaryCluster,
			fmt.Sprintf(config.SetSuperReadonlyClause, "OFF"),
			c.api,
			c.encryptor,
			tidbcloud.BuildClusterProxy,
			config.DefaultClusterProxyTimeout,
		); err != nil {
			return perrors.Annotatef(err, "failed to set super readonly")
		}

		if err = c.RecoverBackup(ctx, r.SecondaryClusterID, r.DetachUserID); err != nil {
			return perrors.Annotatef(err, "failed to recover restore")
		}
	}

	// add SLI of detach time for the cluster
	return db.Transaction(func(tx *gorm.DB) error {
		err = entity.NewReplicationTrackingQuerySet(tx).
			StatusEq(entity.ReplicationStatusDetaching).
			ReplicationIDEq(r.ID).
			One(&track)
		if err != nil {
			return err
		}
		// use the creation of detaching record in history table to record the beginint time of detach
		// use now minus the detaching record `CreatedAt` can calculate the detach durations
		if r.DetachType == entity.DetachTypePlanned {
			metrics.ReplicationPlannedDetachDuration.WithLabelValues(metrics.GetProviderType(r.PrimaryCluster.Free)).Observe(time.Since(track.CreatedAt).Seconds())
		} else {
			metrics.ReplicationForceDetachDuration.WithLabelValues(metrics.GetProviderType(r.PrimaryCluster.Free)).Observe(time.Since(track.CreatedAt).Seconds())
		}
		r.ReplicationStatus = entity.ReplicationStatusDetached
		return c.dao.UpdateReplication(r, []entity.ReplicationDBSchemaField{entity.ReplicationDBSchema.ReplicationStatus})
	})
}

// delete cdc and check the process
func (c *BgContext) OnDeleteCDC(ctx context.Context, r entity.Replication) error {
	var (
		err  error
		task *entity.CDCTask
	)
	if r.CDCTaskID == 0 || r.Task.Status == entity.CDCStatusDeleted || r.Task.Status == entity.CDCStatusDeleting {
		return nil
	}
	task = r.Task
	task.Status = entity.CDCStatusDeleting
	task.Cluster = r.SecondaryCluster
	err = c.cdcdao.UpdateCdcTask(*task, []entity.CDCTaskDBSchemaField{entity.CDCTaskDBSchema.Status})
	if err != nil {
		return err
	}
	return perrors.Errorf("wait for cdc task delete complete")
}

// when the cluster is replicating, we should update the lag time in the record
func (c *BgContext) GetLagTime(ctx context.Context) error {
	var (
		err, errs    error
		lagTime      uint64
		r            entity.Replication
		replications []entity.Replication
	)
	replicatingStatus := entity.ReplicationStatusReplicating
	replications, err = c.dao.FindReplication(dao.ReplicationFilter{
		TenantID: c.tenantAndUser.TenantID.Uint64(),
		Status:   &replicatingStatus,
	})
	if err != nil {
		return err
	}
	for _, r = range replications {
		lagTime, err = c.GetLastDetachLagTime(ctx, &r)
		if err != nil {
			c.log.Debug("failed to get last lag time of replication", zap.Uint64("replication ID", r.ID.Uint64()), zap.Error(err))
			errs = multierr.Append(errs, err)
			continue
		}
		r.LagTimeWhenDetach = &lagTime
		err = db.Transaction(func(tx *gorm.DB) error {
			return c.dao.UpdateReplication(r, []entity.ReplicationDBSchemaField{entity.ReplicationDBSchema.LagTimeWhenDetach})
		})
		errs = multierr.Append(errs, err)
	}
	return errs
}

func (c *BgContext) ScaleSecondary(ctx context.Context) error {
	var (
		err, errs    error
		r            entity.Replication
		replications []entity.Replication
	)
	replicatingStatus := entity.ReplicationStatusReplicating
	replications, err = c.dao.FindReplication(dao.ReplicationFilter{
		TenantID: c.tenantAndUser.TenantID.Uint64(),
		Status:   &replicatingStatus,
	})
	if err != nil {
		return err
	}
	for _, r = range replications {
		err = c.KeepComponentConsistent(ctx, r.PrimaryClusterID, r.SecondaryClusterID)
		if err != nil {
			c.log.Debug("failed to scale secondary cluster", zap.Error(err))
			errs = multierr.Append(errs, err)
		}
	}
	return errs
}

func (c *BgContext) GetLastDetachLagTime(ctx context.Context, r *entity.Replication) (uint64, error) {
	var (
		err           error
		clusterTso    uint64
		checkpointTso uint64
	)
	clusterTso, err = c.GetClusterTSO(ctx, r.PrimaryCluster, tidbcloud.BuildClusterProxy)
	if err != nil {
		return 0, perrors.Annotatef(err, "failed to get cluster tso")
	}
	checkpointTso, err = c.GetCheckPointTSO(ctx, r.SecondaryCluster.ProjectID, r.SecondaryClusterID, r.Task.ID, tidbcloud.BuildClusterProxy)
	if err != nil {
		return 0, perrors.Annotatef(err, "failed to get check point tso")
	}
	return ExtractPhysical(clusterTso) - ExtractPhysical(checkpointTso), nil
}

func ExtractPhysical(tso uint64) uint64 {
	return tso >> config.PhysicalShiftBits
}

// recovering auto backup when detach the replications
func (c *BgContext) RecoverBackup(ctx context.Context, clusterID, userID entity.ID) error {
	var (
		err, createErr error
		backupSchedule entity.BackupSchedule
		cluster        entity.Cluster
	)
	return db.Transaction(func(tx *gorm.DB) error {
		err = entity.NewClusterQuerySet(tx).
			IDEq(clusterID).
			StatusNotIn(entity.DeletedOrClearedClusterStatus...).
			One(&cluster)
		// if active cluster is not exist
		// auto backup schedule should not create for it
		if err == gorm.ErrRecordNotFound {
			return nil
		} else if err != nil {
			return err
		}
		err = entity.NewBackupScheduleQuerySet(tx).
			ClusterIDEq(clusterID).
			TypeEq(entity.BackupType_Auto).
			One(&backupSchedule)
		if err == gorm.ErrRecordNotFound {
			autoBackupConfig := dbutil.GetClusterMaxAutoBackupsConfig(ctx, tx, cluster)
			_, createErr = dbutil.CreateBackupSchedule(tx, userID, clusterID, autoBackupConfig)
			return createErr
		}
		return err
	})
}

func (c BgContext) GetClusterTSO(ctx context.Context, cluster *entity.Cluster, getProxy util.GetProxyFunc) (uint64, error) {
	var (
		err      error
		data     []byte
		cr       *cloudv1.Cluster
		proxy    clusterproxy.ClusterProxy
		ret      constants.MasterStatus
		adminPwd encrypt.Secret
	)

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

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

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

	data, err = proxy.QuerySQL(ctx, adminPwd, config.GetClusterTsoClase)
	if err != nil {
		return 0, perrors.Annotatef(err, "failed to query gc life time")
	}

	err = json.Unmarshal(data, &ret)
	if err != nil {
		return 0, perrors.Annotate(err, "failed to unmarsh get tso model")
	}
	return ret.Position[0], nil
}

func (c BgContext) GetCheckPointTSO(ctx context.Context, projectID, clusterID, taskID entity.ID, getProxy util.GetProxyFunc) (uint64, error) {
	var (
		err     error
		cr      *cloudv1.Cluster
		proxy   clusterproxy.ClusterProxy
		cdcList []clusterproxy.CdcInfo
	)
	cr, err = c.api.GetCluster(ctx, projectID.Uint64(), clusterID.Uint64())
	if err != nil {
		return 0, err
	}

	proxy, err = getProxy(clusterID.Uint64(), cr.Status, config.DefaultClusterProxyTimeout)
	if err != nil {
		return 0, err
	}

	cdcList, err = proxy.GetCDCTaskInfo(ctx)
	if err != nil {
		return 0, err
	}

	for _, cdc := range cdcList {
		if cdc.ID == cdcModel.CdcChangefeedID(taskID.Uint64()) {
			return cdc.CpTso, nil
		}
	}
	return 0, perrors.Errorf("failed to found changefeed")

}

// for the normal secondary cluster, it should keep consistancy with the primary one
// this only running when the replication is working and not detach, the cluster should scale
// the replicating clluster can make sure keep asstistant by the interface
// this function is cover the corner case, and even if the cluster scale when it is in creating
func (c BgContext) KeepComponentConsistent(ctx context.Context, pClusterID, sClusterID entity.ID) error {
	logger := log.WithContextS(ctx).With(
		zap.Uint64("pClusterId", pClusterID.Uint64()),
		zap.Uint64("sClusterId", sClusterID.Uint64()))
	var (
		err      error
		pCluster entity.Cluster
		sCluster entity.Cluster
	)
	return db.Transaction(func(tx *gorm.DB) error {
		err = entity.NewClusterQuerySet(dbutil.PreloadCluster(tx)).
			IDEq(pClusterID).
			One(&pCluster)
		if err != nil {
			return err
		}
		err = entity.NewClusterQuerySet(dbutil.PreloadCluster(tx)).
			IDEq(sClusterID).
			One(&sCluster)
		if err != nil {
			return err
		}
		if pCluster.Status.Creating() {
			logger.Infof("the primary cluster is creating, wait next time")
			return nil
		}
		if sCluster.Status == entity.ClusterStatusScaling {
			logger.Infof("the secondary cluster is scaling, wait next time")
			return nil
		}

		same := dbutil.IsComponentsReplicaSame(pCluster.ComponentsV2, sCluster.ComponentsV2)
		if same {
			logger.Infof("primary and secondary clusters are consistent, no need to scale")
			return nil
		}

		// scale the secondary cluster to keep the consistency
		logger.Infof("try to scale secondary, newComponents=%+v", pCluster.ComponentsV2)
		components := make([]*core.ClusterComponent, 0, len(pCluster.ComponentsV2))
		for _, c := range pCluster.ComponentsV2 {
			component := cvt.ClusterComponentModel2Core(c)
			components = append(components, component)
		}
		scaleClusterReq := &core.ScaleClusterReq{
			ClusterId:           sClusterID.Uint64(),
			NewComponents:       components,
			AllowScaleIn:        true,
			AllowScaleSecondary: true,
			BaseReq:             nil,
		}
		_, err = grpcli.ScaleCluster(ctx, scaleClusterReq)
		if err != nil {
			return err
		}
		return nil
	})
}

func (c BgContext) ensureTiCDCDeleted(ctx context.Context, cluster *entity.Cluster) error {
	var (
		err error
	)

	_, err = c.api.GetTicdc(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
	if err != nil && !apierrors.IsNotFound(err) {
		return err
	} else if err != nil && apierrors.IsNotFound(err) {
		return nil
	}

	return perrors.Errorf("ticdc of cluster %d is not deleted", cluster.ID.Uint64())
}