package cdc

import (
	"context"
	"fmt"
	"net"
	"net/url"
	"strings"
	"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"
	replicationDao "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/handler/billing2"
	api "github.com/pingcap/dbaas/central/infra-api"
	"github.com/pingcap/dbaas/central/metrics"
	"github.com/pingcap/dbaas/central/model/billing"
	"github.com/pingcap/dbaas/central/model/cluster/clusterproxy/tidbcloud"
	"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/clusterproxy"
	"github.com/tidbcloud/pkgv2/secret/encrypt"
	"go.uber.org/zap"
	"gorm.io/gorm"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	"k8s.io/utils/pointer"
)

// Model is created with NewModel
type Model struct {
	dao            cdcDao.CDCDaoFacade
	log            *zap.Logger
	api            api.InfraFacade
	encryptor      encrypt.Encryptor
	clusterdao     clusterDao.ClusterDaoFacade
	replicationdao replicationDao.ReplicationDaoFacade
	billingModel   billing.Model
}

// NewModel returns a Model
func NewModel(
	log *zap.Logger,
	encryptor encrypt.Encryptor,
	infraAPI api.InfraFacade,
	dao cdcDao.CDCDaoFacade,
	clusterdao clusterDao.ClusterDaoFacade,
	replicationdao replicationDao.ReplicationDaoFacade,
	billingModel billing.Model) Model {
	return Model{
		log:            log,
		encryptor:      encryptor,
		api:            infraAPI,
		dao:            dao,
		clusterdao:     clusterdao,
		replicationdao: replicationdao,
		billingModel:   billingModel,
	}
}

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

var (
	DisplayNameTransfer = map[entity.CDCDownStreamType]string{
		entity.CDCDownStreamTypeKafka: "Kafka",
		entity.CDCDownStreamTypeMysql: "MySQL",
		entity.CDCDownStreamTypeTiDB:  "TiDB",
	}

	//tidbSemVersionConstraint, _ = semver.NewConstraint(">= v5.3.0")
)

const (
	GetCDCUrlPath = "api/v1/changefeeds"
	// default message from cdc doc
	// See: https://docs.pingcap.com/zh/tidb/stable/manage-ticdc
	DefaultMessageBytes = "67108864"
	DefaultProtocol     = entity.KafkaProtocolTypeCanalJson
	DefaultWorkCount    = 16
	DefaultMaxTxnRow    = 256
	// version is not used in kafka client now
	// next version of cdc will get the kafka version automatically
	DefaultVersion = "2.4.0"

	// machine and replicas will be over write by infra-api
	DefaultAWSCDCInstance = "c5.9xlarge"
	DefaultGCPCDCInstance = "n1-standard-32"

	// const error messages
	// TODO(yinliang): maybe use error code and save a error message list between frontend and backend for error messages for display
	RecoveringErr            = "The cluster is being recovering. Cannot create connector during `Recovering` process."
	ImportingErr             = "The cluster is being imported. Cannot create connector during `Importing` process."
	CreatingErr              = "The cluster is being creating. Cannot create connector during `Creating` process."
	FailedToConnectErr       = "Cannot connect to the %s Address"
	FailedToGetTSO           = "Cannot get the TSO in TiDB Cluster"
	FailedToGetChangefeedErr = "Cannot get changefeed, may have been deleted by mistake"
	FailedToGetConnectorErr  = "Connector has been deleted"
	FailedToGetProjectCIDR   = "Cannot get project cidr of cluster"
	NotallownProjectCIDRErr  = "it is not allowed to sink project cidr downstream"
	NotallowDevTierErr       = "it is not allowed to create connector for dev tier clusters"
	NotallowReplicationErr   = "it is not allowed to create connector when replicating"

	CDCStatusNormal  = "normal"
	CDCStatusError   = "error"
	CDCStatusFailed  = "failed"
	CDCStatusStopped = "stopped"
)

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

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

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)
}

// BgContext is meant for the system worker/controller
type BgContext struct {
	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{
		UserContext: UserContext{
			Model:         m,
			tenantAndUser: &tenant.User{User: systemUser, TenantID: tenantID},
			log:           m.log.With(zap.Uint64("tenantID", tenantID.Uint64()), zap.Uint64("userID", uint64(systemUser.ID)), zap.String("module", "cdc")),
		},
	}, nil
}

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

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

// manage the life cycle of cdc changefeed
func (c BgContext) ManageCdcChangefeed(ctx context.Context, getProxy util.GetProxyFunc) error {
	var (
		err                   error
		ignoreCheckchangefeed bool
		tenantID              entity.ID
		cdcRecords            []entity.CDCTask
		clusters              []entity.Cluster
		replications          []entity.Replication
		secondaryCluster      *entity.Cluster
	)
	tenantID = c.tenantAndUser.TenantID
	clusters, err = c.clusterdao.FindClusters(dbutil.ClusterFilter{
		TenantID: &tenantID,
		Free:     pointer.BoolPtr(false),
		Statuss: []entity.ClusterStatus{
			entity.ClusterStatusNormal,
			entity.ClusterStatusScaling,
			entity.ClusterStatusDeleted},
	})
	if err != nil {
		return errors.Annotatef(err, "failed to fetch clusters")
	}
	for _, cluster := range clusters {
		cdcRecords, err = c.dao.FindCDCTasks(cdcDao.CdcFilter{
			Deleted:   pointer.BoolPtr(false),
			ClusterID: cluster.ID.Uint64(),
		})
		if err != nil {
			return errors.Annotatef(err, "failed to fetch cdc tasks")
		}
		for _, r := range cdcRecords {
			// check cdc cluster status, create or wait for available
			// check whether the cdc cluster is belong to replication
			// if so the ticdc cluster is deploy in the downstream of replication
			// ensure the ticdc cluster in the downstream tidb cluster
			replications, err = c.replicationdao.FindReplication(replicationDao.ReplicationFilter{
				TenantID: c.tenantAndUser.TenantID.Uint64(),
				CdcID:    r.ID.Uint64(),
			})
			if err != nil {
				c.Log().Error("failed to fetch replication list", zap.Uint64("task ID", r.ID.Uint64()), zap.Error(err))
				continue
			}
			if len(replications) == 0 && cluster.ReplicationID != nil {
				c.Log().Error("wait for replication update", zap.Uint64("task ID", r.ID.Uint64()), zap.Error(err))
				continue
			}
			secondaryCluster = nil
			if len(replications) != 0 {
				// it means cdc cluster deploy in the downstream tidb cluster
				cluster = *replications[0].PrimaryCluster
				secondaryCluster = replications[0].SecondaryCluster
				ignoreCheckchangefeed = (cluster.Status == entity.ClusterStatusDeleted ||
					(replications[0].ReplicationStatus == entity.ReplicationStatusDetaching && replications[0].DetachType == entity.DetachTypeNone))
			} else {
				ignoreCheckchangefeed = (cluster.Status == entity.ClusterStatusDeleted)
			}
			// if cluster is deleted, ignore the cdc status
			if !ignoreCheckchangefeed {
				if err = c.ensureTiCdcCluster(ctx, &cluster, secondaryCluster); err != nil {
					c.Log().Error("failed to ensure cdc cluster", zap.Uint64("task ID", r.ID.Uint64()), zap.Error(err))
					continue
				}
			}
			if err = c.manageCdcRecord(ctx, ignoreCheckchangefeed, r, secondaryCluster, getProxy); err != nil {
				c.Log().Error("failed to deal with cdc task", zap.Uint64("task ID", r.ID.Uint64()), zap.Error(err))
				continue
			}
		}
	}

	return nil
}

func (c BgContext) manageCdcRecord(
	ctx context.Context,
	ignoreCheckchangefeed bool,
	r entity.CDCTask,
	secondaryCluster *entity.Cluster,
	getProxy util.GetProxyFunc) error {
	var (
		find         bool
		err          error
		sink         string
		cluster      *entity.Cluster
		cdcInfoList  []clusterproxy.CdcInfo
		replications []entity.Replication
	)

	cluster = r.Cluster
	if secondaryCluster != nil {
		cluster = secondaryCluster
		replications, err = c.replicationdao.FindReplication(replicationDao.ReplicationFilter{
			TenantID: c.tenantAndUser.TenantID.Uint64(),
			CdcID:    r.ID.Uint64(),
		})
		if err != nil {
			c.Log().Error("failed to fetch replication list", zap.Uint64("task ID", r.ID.Uint64()), zap.Error(err))
		}
	}

	// if cluster is deleted, ignore the real task
	// meantime set all tasks as deleted
	if ignoreCheckchangefeed {
		r.Status = entity.CDCStatusDeleted
		err = c.dao.UpdateCdcTask(r, []entity.CDCTaskDBSchemaField{entity.CDCTaskDBSchema.Status})
		if err != nil {
			return errors.Annotatef(err, "failed to update cdc task status")
		}
		return nil
	}

	cdcInfoList, err = c.getCdcList(ctx, cluster, getProxy)
	if err != nil {
		return errors.Annotatef(err, "failed to get cdc list")
	}

	find = false
	for _, cdcInfo := range cdcInfoList {
		if CdcChangefeedID(r.ID.Uint64()) == cdcInfo.ID {
			find = true
			// if record is deleting, try to delete changefeed first, force detach can skip delete changefeed
			if r.Status == entity.CDCStatusDeleting {
				err = c.deleteCdcTask(ctx, &r, secondaryCluster, getProxy)
				if err != nil {
					if len(replications) > 0 && entity.DetachTypeForce == replications[0].DetachType {
						c.Log().Info("force detach delete changefeed failed, skip it", zap.Uint64("task ID", r.ID.Uint64()), zap.Error(err))
						r.Status = entity.CDCStatusDeleted
						err = c.dao.UpdateCdcTask(r, []entity.CDCTaskDBSchemaField{entity.CDCTaskDBSchema.Status})
						if err != nil {
							return errors.Annotatef(err, "failed to update cdc task status of force detach")
						}
					} else {
						return errors.Annotatef(err, "failed to delete cdc task")
					}
				}
			} else {
				// if record in normal/failed/creating status, change the cdc task status in time
				if cdcInfo.State == CDCStatusNormal {
					r.Status = entity.CDCStatusNormal
					//TODO(yinliang) cdc will still show error message after status back to normal
					// workaround this scene
					r.ErrorMessage = ""
					err = c.dao.UpdateCdcTask(r, []entity.CDCTaskDBSchemaField{entity.CDCTaskDBSchema.Status, entity.CDCTaskDBSchema.ErrorMessage})
					if err != nil {
						return errors.Annotatef(err, "failed to upadte cdc task")
					}
				} else if cdcInfo.State == CDCStatusError || cdcInfo.State == CDCStatusFailed {
					r.Status = entity.CDCStatusFailed
					r.ErrorMessage = cdcInfo.Error.Message
					err = c.dao.UpdateCdcTask(r, []entity.CDCTaskDBSchemaField{entity.CDCTaskDBSchema.Status, entity.CDCTaskDBSchema.ErrorMessage})
					if err != nil {
						return errors.Annotatef(err, "failed to upadte cdc task")
					}
				}
			}
		}
	}
	if !find {
		if r.Status == entity.CDCStatusDeleting {
			r.Status = entity.CDCStatusDeleted
			err = c.dao.UpdateCdcTask(r, []entity.CDCTaskDBSchemaField{entity.CDCTaskDBSchema.Status})
			if err != nil {
				return errors.Annotatef(err, "failed to update cdc task status")
			}
		} else if r.Status == entity.CDCStatusCreating {
			// create new changefeed
			sink, err = c.assembleSinkUrl(&r)
			if err != nil {
				return errors.Annotatef(err, "failed to assemble sink")
			}
			err = c.createCdcTask(ctx, &r, secondaryCluster, sink, getProxy)
			if err != nil {
				metrics.ChangefeedCreateResultCounter.WithLabelValues(metrics.GetProviderType(r.Cluster.Free), "failed").Inc()
				return errors.Annotatef(err, "failed to create cdc task")
			}
			metrics.ChangefeedCreateResultCounter.WithLabelValues(metrics.GetProviderType(r.Cluster.Free), "succeed").Inc()
		} else {
			r.Status = entity.CDCStatusFailed
			r.ErrorMessage = FailedToGetChangefeedErr
			err = c.dao.UpdateCdcTask(r, []entity.CDCTaskDBSchemaField{entity.CDCTaskDBSchema.Status, entity.CDCTaskDBSchema.ErrorMessage})
			if err != nil {
				return errors.Annotatef(err, "failed to upadte cdc task")
			}
		}
	}
	return nil
}

func (c BgContext) assembleSinkUrl(task *entity.CDCTask) (string, error) {
	var (
		err          error
		sqlSink      string
		kafkaSink    string
		realPassword encrypt.Secret
		sqlTask      *entity.CDCSQLTask
		kafkaTask    *entity.CDCKafkaTask
	)
	if task.DownStreamType == entity.CDCDownStreamTypeKafka {
		kafkaTask, err = c.dao.GetCdcKafkaTask(task.ID)
		if err != nil {
			return "", errors.Annotatef(err, "failed to get kafka task")
		}
		kafkaSink = fmt.Sprintf("kafka://%s:%s/%s?kafka-version=%s&max-message-bytes=%s&protocol=%s",
			kafkaTask.DownStreamEndpoint,
			kafkaTask.DownStreamPort,
			kafkaTask.Topic,
			kafkaTask.Version,
			kafkaTask.MaxMessageBytes,
			kafkaTask.Protocol)
		return kafkaSink, nil
	}
	// other's are sql type cdc protocol
	sqlTask, err = c.dao.GetCdcSqlTask(task.ID)
	if err != nil {
		return "", errors.Annotatef(err, "failed to get sql task")
	}

	realPassword, err = encrypt.DecryptBase64(c.encryptor, sqlTask.Password)
	if err != nil {
		return "", errors.Annotatef(err, "failed to decrypted password")
	}
	sqlSink = fmt.Sprintf("%s://%s:%s@%s:%s?worker-count=%d&max-txn-row=%d",
		task.DownStreamType,
		sqlTask.User,
		url.QueryEscape(string(realPassword)),
		sqlTask.DownStreamEndpoint,
		sqlTask.DownStreamPort,
		sqlTask.WorkCount,
		sqlTask.MaxTxnRow)
	return sqlSink, nil
}

func (c BgContext) ensureTiCdcCluster(ctx context.Context, cluster *entity.Cluster, secondaryCluster *entity.Cluster) error {
	var (
		err                                error
		clusterID, projectID               uint64
		cdc                                *cloudv1.TiCDC
		clusterResource, secondaryResource *cloudv1.Cluster
	)
	if secondaryCluster == nil {
		clusterID = cluster.ID.Uint64()
		projectID = cluster.ProjectID.Uint64()
	} else {
		clusterID = secondaryCluster.ID.Uint64()
		projectID = secondaryCluster.ProjectID.Uint64()
	}
	cdc, err = c.api.GetTicdc(ctx, projectID, clusterID)
	if err != nil && !apierrors.IsNotFound(err) {
		return errors.Annotatef(err, "failed to get ticdc")
	} else if apierrors.IsNotFound(err) {
		clusterResource, err = c.api.GetCluster(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
		if err != nil {
			return errors.Annotatef(err, "failed to get cluster")
		}
		if secondaryCluster == nil {
			cdc = c.assembleTiCdc(clusterResource.Name, clusterResource.Spec.NetworkName, "", clusterResource.GetProviderType())
		} else {
			secondaryResource, err = c.api.GetCluster(ctx, secondaryCluster.ProjectID.Uint64(), secondaryCluster.ID.Uint64())
			if err != nil {
				return errors.Annotatef(err, "failed to get secondary cluster")
			}
			cdc = c.assembleTiCdc(secondaryResource.Name, secondaryResource.Spec.NetworkName, clusterResource.Name, secondaryResource.GetProviderType())
		}
		err = c.api.CreateTicdcIgnoreExist(ctx, projectID, clusterID, cdc)
		if err != nil {
			return errors.Annotatef(err, "failed to create cluster")
		}
		return errors.Errorf("wait for cdc cluster created")
	}

	if cdc.Status.Phase == cloudv1.TiCDCPhaseDeleting {
		return errors.Errorf("wait for cdc cluster deleting")
	} else if cdc.Status.Phase != cloudv1.TiCDCPhaseRunning {
		return errors.Errorf("wait for cdc running now %s", cdc.Status.Phase)
	}

	return nil
}

func (c BgContext) assembleTiCdc(crName, networkName, upstreamcrName, providerType string) *cloudv1.TiCDC {
	// cdc server configuration will be overwritten in infra-api layer
	// here is test configuration
	var instanceType string
	if providerType == "aws" {
		instanceType = DefaultAWSCDCInstance
	} else {
		instanceType = DefaultGCPCDCInstance
	}
	return &cloudv1.TiCDC{
		Spec: cloudv1.TiCDCSpec{
			InstanceType:        instanceType,
			Replicas:            1,
			StorageSize:         *resource.NewQuantity(1024*1024*1024*1024, resource.DecimalSI),
			NetworkName:         networkName,
			ClusterName:         crName,
			UpstreamClusterName: upstreamcrName,
		},
	}
}

func CreateCdcTaskImpl(
	clusterID entity.ID,
	connectType central.ConnectorType,
	url string,
	topic string,
	user string,
	password string,
	startTS uint64,
	clusterdao clusterDao.ClusterDaoFacade,
	enc encrypt.Encryptor,
	log *zap.Logger,
	ignoreReplication bool) (*central.ConnectorID, error) {
	var (
		err               error
		domain            string
		port              string
		encryptedPassword string
		epList            []string
		task              entity.CDCTask
		sqlTask           entity.CDCSQLTask
		kafkaTask         entity.CDCKafkaTask
		cType             entity.CDCDownStreamType
		cluster           entity.Cluster
	)

	cluster, err = clusterdao.GetCluster(clusterID)
	if err != nil {
		return nil, errors.Annotatef(err, "failed to get cluster")
	}
	if cluster.Status == entity.ClusterStatusRecovering {
		return nil, errors.Errorf(RecoveringErr)
	} else if cluster.ImportID != nil {
		return nil, errors.Errorf(ImportingErr)
	} else if cluster.Status == entity.ClusterStatusInitializing ||
		cluster.Status == entity.ClusterStatusPreparing ||
		cluster.Status == entity.ClusterStatusLaunching {
		return nil, errors.Errorf(CreatingErr)
	}

	if cluster.IsDevTier() {
		return nil, errors.New(NotallowDevTierErr)
	}
	if connectType != central.ConnectorType_TIDB && cluster.ReplicationID != nil {
		return nil, errors.New(NotallowReplicationErr)
	}

	if connectType == central.ConnectorType_KAFKA {
		cType = entity.CDCDownStreamTypeKafka
	} else if connectType == central.ConnectorType_TIDB {
		cType = entity.CDCDownStreamTypeTiDB
	} else if connectType == central.ConnectorType_MYSQL {
		cType = entity.CDCDownStreamTypeMysql
	} else {
		return nil, errors.Errorf("unknown cdc downstream type")
	}

	// endpoint list have been validate in service layer
	epList = strings.Split(url, ":")
	domain = epList[0]
	port = epList[1]

	if cType == entity.CDCDownStreamTypeKafka {
		err = db.Transaction(func(tx *gorm.DB) error {
			task = entity.CDCTask{
				ClusterID:         clusterID,
				DownStreamType:    cType,
				Status:            entity.CDCStatusCreating,
				IgnoreReplication: ignoreReplication,
			}
			return task.Create(tx)
		})
		if err != nil {
			return nil, errors.Annotatef(err, "failed to create cdc task")
		}

		err = db.Transaction(func(tx *gorm.DB) error {
			kafkaTask = entity.CDCKafkaTask{
				CDCTaskID:          task.ID,
				DownStreamEndpoint: domain,
				DownStreamPort:     port,
				Topic:              topic,
				Version:            DefaultVersion,
				MaxMessageBytes:    DefaultMessageBytes,
				Protocol:           DefaultProtocol,
			}
			return kafkaTask.Create(tx)
		})
		if err != nil {
			return nil, errors.Annotatef(err, "failed to create cdc kafka task")
		}
		return &central.ConnectorID{
			ConnectorId: task.ID.Uint64(),
		}, nil
	}
	err = db.Transaction(func(tx *gorm.DB) error {
		task = entity.CDCTask{
			ClusterID:         clusterID,
			DownStreamType:    cType,
			Status:            entity.CDCStatusCreating,
			IgnoreReplication: ignoreReplication,
			StartTimeStamp:    startTS,
		}
		return task.Create(tx)
	})
	if err != nil {
		return nil, errors.Annotatef(err, "failed to create cdc task")
	}

	encryptedPassword, err = encrypt.EncryptBase64(enc, encrypt.Secret(password))
	if err != nil {
		log.Error("failed to encrypt password", zap.Error(err))
		return nil, errors.Annotatef(err, "failed to encrypt password")
	}

	err = db.Transaction(func(tx *gorm.DB) error {
		sqlTask = entity.CDCSQLTask{
			CDCTaskID:          task.ID,
			DownStreamEndpoint: domain,
			DownStreamPort:     port,
			User:               user,
			Password:           encryptedPassword,
			WorkCount:          DefaultWorkCount,
			MaxTxnRow:          DefaultMaxTxnRow,
		}
		return sqlTask.Create(tx)
	})
	if err != nil {
		return nil, errors.Annotatef(err, "failed to create cdc kafka task")
	}
	return &central.ConnectorID{
		ConnectorId: task.ID.Uint64(),
	}, nil
}

// create cdc task
func (c UserContext) CreateCdcTask(ctx context.Context, clusterID, projectID entity.ID, connectType central.ConnectorType, url string, topic string, user string, password string, startTS uint64, ignoreReplication bool) (*central.ConnectorID, error) {
	// TODO: (yinliang) refactor test codes
	var (
		err error
	)
	if err = c.TestCdcTask(ctx, projectID, clusterID, connectType, url, topic, user, password, startTS, tidbcloud.BuildClusterProxy); err != nil {
		return nil, err
	}

	return CreateCdcTaskImpl(
		clusterID,
		connectType,
		url,
		topic,
		user,
		password,
		startTS,
		c.clusterdao,
		c.encryptor,
		c.log,
		ignoreReplication)
}

func (c UserContext) UpdateClusterTCU(clusterID entity.ID, tcuNumber uint64) error {
	cluster, err := c.clusterdao.GetCluster(clusterID)
	if err != nil {
		return errors.Annotatef(err, "failed to get cluster")
	}
	cluster.TCUNumber = tcuNumber

	return c.clusterdao.UpdateCluster(cluster, []entity.ClusterDBSchemaField{entity.ClusterDBSchema.TCUNumber}, false)
}

// delete cdc task
func (c UserContext) DeleteCdcTask(taskID, clusterID, projectID entity.ID) error {
	var (
		err  error
		task *entity.CDCTask
	)
	task, err = c.dao.GetCdcTaskByID(taskID, clusterID, projectID)
	if err != nil {
		return errors.Annotatef(err, "failed to get cdc task")
	}
	task.Status = entity.CDCStatusDeleting
	return c.dao.UpdateCdcTask(*task, []entity.CDCTaskDBSchemaField{entity.CDCTaskDBSchema.Status})
}

// test connect to mysql/kafka, and check weather the parameter is valid
func (c UserContext) TestCdcTask(
	ctx context.Context,
	projectID entity.ID,
	clusterID entity.ID,
	connectType central.ConnectorType,
	url string,
	topic string,
	user string,
	password string,
	startTS uint64,
	getProxy util.GetProxyFunc) error {
	var (
		err       error
		cType     entity.CDCDownStreamType
		cr        *cloudv1.Cluster
		proxy     clusterproxy.ClusterProxy
		safepoint *clusterproxy.Safepoint
	)

	if connectType == central.ConnectorType_KAFKA {
		cType = entity.CDCDownStreamTypeKafka
	} else if connectType == central.ConnectorType_TIDB {
		cType = entity.CDCDownStreamTypeTiDB
	} else {
		cType = entity.CDCDownStreamTypeMysql
	}

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

	projectNetwork, err := c.clusterdao.GetProjectCIDRByClusterID(projectID, clusterID)
	if err != nil && err != gorm.ErrRecordNotFound {
		return errors.Errorf(FailedToGetProjectCIDR)
	}
	if projectNetwork.CIDR == "" {
		return errors.Errorf(FailedToGetProjectCIDR)
	}

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

	// if the test connection type is mysql
	// start time stamp need to compare with gc safe point tso
	if (connectType == central.ConnectorType_MYSQL || connectType == central.ConnectorType_TIDB) && startTS != 0 {
		safepoint, err = proxy.GetPDSafepoint(ctx)
		if err != nil {
			return err
		}

		if safepoint.SP >= startTS {
			return errors.Errorf(FailedToGetTSO)
		}
	}
	if connectType == central.ConnectorType_MYSQL || connectType == central.ConnectorType_KAFKA {
		urlList := strings.Split(url, ":")
		ip := net.ParseIP(urlList[0])
		_, netCidr, err := net.ParseCIDR(projectNetwork.CIDR)
		if err == nil && netCidr.Contains(ip) {
			return errors.Errorf(NotallownProjectCIDRErr)
		}
	}

	err = proxy.TestConnection(ctx,
		string(cType),
		url,
		user,
		password,
		topic)
	if err != nil {
		return errors.Annotatef(err, fmt.Sprintf(FailedToConnectErr, DisplayNameTransfer[cType]))
	}
	return nil
}

func (c UserContext) GetCdcTask(taskID, clusterID, projectID entity.ID) (*central.Connector, error) {
	var (
		err  error
		task *entity.CDCTask
	)
	task, err = c.dao.GetCdcTaskByID(taskID, clusterID, projectID)
	if err != nil {
		return nil, errors.Annotatef(err, "failed to get cdc task")
	}
	if task.Status == entity.CDCStatusDeleting || task.Status == entity.CDCStatusDeleted {
		return nil, errors.New(FailedToGetConnectorErr)
	}
	return c.assembleCdc(task)
}

func (c UserContext) ListCdcTask(clusterID entity.ID, connectType central.ConnectorType) (*central.Connectors, error) {
	var (
		err   error
		tasks []entity.CDCTask
		item  *central.Connector
		items []*central.Connector
		cType entity.CDCDownStreamType
		qs    entity.CDCTaskQuerySet
	)

	cType = entity.CDCDownStreamTypeAll
	if connectType == central.ConnectorType_KAFKA {
		cType = entity.CDCDownStreamTypeKafka
	} else if connectType == central.ConnectorType_TIDB {
		cType = entity.CDCDownStreamTypeTiDB
	} else if connectType == central.ConnectorType_MYSQL {
		cType = entity.CDCDownStreamTypeMysql
	}

	err = db.Transaction(func(tx *gorm.DB) error {
		qs = entity.NewCDCTaskQuerySet(tx).ClusterIDEq(clusterID)
		if cType != entity.CDCDownStreamTypeAll {
			qs.DownStreamTypeEq(cType)
		}
		return qs.All(&tasks)
	})
	if err != nil {
		return nil, errors.Annotatef(err, "failed to get tasks")
	}
	for _, t := range tasks {
		if t.Status == entity.CDCStatusDeleting || t.Status == entity.CDCStatusDeleted {
			continue
		}
		item, err = c.assembleCdc(&t)
		if err != nil {
			c.log.Error("failed to assemble task", zap.Uint64("taskid", t.ID.Uint64()), zap.Error(err))
			continue
		}
		items = append(items, item)
	}

	return &central.Connectors{
		Total: uint32(len(items)),
		Items: items,
	}, nil
}

func (c UserContext) assembleCdc(task *entity.CDCTask) (*central.Connector, error) {
	var (
		err           error
		kafkaTask     entity.CDCKafkaTask
		sqlTask       entity.CDCSQLTask
		connectorType central.ConnectorType
		status        central.Connector_Status
	)

	if task.DownStreamType == entity.CDCDownStreamTypeKafka {
		connectorType = central.ConnectorType_KAFKA
	} else if task.DownStreamType == entity.CDCDownStreamTypeTiDB {
		connectorType = central.ConnectorType_TIDB
	} else {
		connectorType = central.ConnectorType_MYSQL
	}

	if task.Status == entity.CDCStatusCreating {
		status = central.Connector_CREATING
	} else if task.Status == entity.CDCStatusFailed {
		status = central.Connector_INVALID
	} else {
		status = central.Connector_NORMAL
	}

	if task.DownStreamType == entity.CDCDownStreamTypeKafka {
		err = db.Transaction(func(tx *gorm.DB) error {
			return entity.NewCDCKafkaTaskQuerySet(tx).
				CDCTaskIDEq(task.ID).
				One(&kafkaTask)
		})
		if err != nil {
			return nil, errors.Annotatef(err, "failed to get kafka task")
		}
		return &central.Connector{
			ConnectorId:   task.ID.Uint64(),
			Url:           fmt.Sprintf("%s:%s", kafkaTask.DownStreamEndpoint, kafkaTask.DownStreamPort),
			Topic:         kafkaTask.Topic,
			ConnectorType: connectorType,
			Status:        status,
			Error:         task.ErrorMessage,
		}, nil
	}
	err = db.Transaction(func(tx *gorm.DB) error {
		return entity.NewCDCSQLTaskQuerySet(tx).
			CDCTaskIDEq(task.ID).
			One(&sqlTask)
	})
	if err != nil {
		return nil, errors.Annotatef(err, "failed to get sql task")
	}
	return &central.Connector{
		ConnectorId:   task.ID.Uint64(),
		Url:           fmt.Sprintf("%s:%s", sqlTask.DownStreamEndpoint, sqlTask.DownStreamPort),
		User:          sqlTask.User,
		ConnectorType: connectorType,
		Status:        status,
		Error:         task.ErrorMessage,
	}, nil
}

func (c BgContext) createCdcTask(
	ctx context.Context,
	cdc *entity.CDCTask,
	secondaryCluster *entity.Cluster,
	sink string,
	getProxy util.GetProxyFunc) error {
	var (
		err     error
		cr      *cloudv1.Cluster
		cluster *entity.Cluster
		proxy   clusterproxy.ClusterProxy
	)
	cluster = cdc.Cluster
	if secondaryCluster != nil {
		cluster = secondaryCluster
	}

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

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

	err = proxy.CreateCDCTask(ctx, sink, cdc.StartTimeStamp, CdcChangefeedID(cdc.ID.Uint64()), cdc.IgnoreReplication)
	if err != nil {
		return errors.Annotatef(err, fmt.Sprintf(FailedToConnectErr, DisplayNameTransfer[cdc.DownStreamType]))
	}
	return nil
}

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

	cluster = cdc.Cluster
	if secondaryCluster != nil {
		cluster = secondaryCluster
	}

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

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

	return proxy.DeleteCDCTask(ctx, CdcChangefeedID(cdc.ID.Uint64()))
}

func (c BgContext) getCdcList(ctx context.Context, cluster *entity.Cluster, getProxy util.GetProxyFunc) ([]clusterproxy.CdcInfo, error) {
	cr, err := c.api.GetCluster(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
	if err != nil {
		return nil, err
	}

	proxy, err := getProxy(cluster.ID.Uint64(), cr.Status, config.DefaultClusterProxyTimeout)
	if err != nil {
		return nil, err
	}
	return proxy.GetCDCTaskInfo(ctx)
}

func CdcChangefeedID(taskID uint64) string {
	return fmt.Sprintf("changefeed-%d", taskID)
}

// clean up cdc machine and k8s reources if no changefeed exist
func (c BgContext) CleanUnusedCdcCluster(ctx context.Context) error {
	var (
		err        error
		tenantID   entity.ID
		clusters   []entity.Cluster
		cdcRecords []entity.CDCTask
	)
	tenantID = c.tenantAndUser.TenantID
	clusters, err = c.clusterdao.FindClusters(dbutil.ClusterFilter{
		TenantID: &tenantID,
		Free:     pointer.BoolPtr(false),
		Statuss: []entity.ClusterStatus{
			entity.ClusterStatusNormal,
			entity.ClusterStatusScaling,
			entity.ClusterStatusDeleted},
	})
	if err != nil {
		return errors.Annotatef(err, "failed to fetch clusters records")
	}
	for _, cluster := range clusters {
		cdcRecords, err = c.dao.FindCDCTasks(cdcDao.CdcFilter{Deleted: pointer.BoolPtr(false), ClusterID: cluster.ID.Uint64()})
		if err != nil {
			c.log.Error("failed to get cdc tasks list", zap.Error(err))
			continue
		}
		if len(cdcRecords) == 0 {
			c.Log().Info("cdc records is zero", zap.Uint64("cluster ID", cluster.ID.Uint64()))
			// delete cdc machine and k8s resources if no cdc task exists
			if err = c.api.DeleteTicdcIfExist(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64()); err != nil {
				c.log.Error("failed to delete cdc resources", zap.Error(err))
				continue
			}
		}
	}
	return nil
}

func getMeteringTimeStamp() time.Time {
	now := time.Now()
	return now.Truncate(time.Hour)
}

func getTCU(cluster entity.Cluster) uint64 {
	if cluster.TCUNumber != 0 {
		return cluster.TCUNumber
	}
	mappingTable := config.Global().BillingConfig.CDCTCUMappingTable

	components := cluster.ComponentsV2
	var tcu uint64 = 0
	for _, component := range components {
		if component.ComponentType != entity.TiKV {
			continue
		}
		replicas := int(component.Replicas) * int(component.InternalProfile.InfraCompute.CPUNum)
		for _, mappingItem := range mappingTable {
			if replicas >= mappingItem[0] {
				tcu = uint64(mappingItem[1])
			}
		}
	}
	return tcu
}

func (c UserContext) GetClusterTCU(clusterID entity.ID) (uint64, error) {
	cluster, err := c.clusterdao.GetCluster(clusterID)
	if err != nil {
		return 0, errors.Annotatef(err, "failed to get cluster")
	}
	cdcRecords, err := c.dao.FindCDCTasks(cdcDao.CdcFilter{Deleted: pointer.BoolPtr(false), ClusterID: cluster.ID.Uint64(), Status: entity.CDCStatusNormal})
	if err != nil {
		return 0, errors.Annotatef(err, "failed to get cdc tasks list")
	}
	if len(cdcRecords) == 0 {
		return 0, nil
	}
	return getTCU(cluster), nil
}

const TCUCostUnitPattern = "ticdc-tcu:%s:%s"
const HourInMinutes = 60

func (c BgContext) createRecordAndSendMetering(cluster entity.Cluster) error {
	tenantID := cluster.TenantID
	costTime := getMeteringTimeStamp()
	costID := fmt.Sprintf("cdc-tcu-%d-%d-%s", tenantID, cluster.ID, costTime.Format("2006010215"))
	tcuNumber := getTCU(cluster)
	record := entity.CDCTCUMeteringRecord{
		CostID:    costID,
		Status:    entity.CDCMeteringStatusInit,
		CostTime:  costTime,
		TenantID:  uint64(tenantID),
		ClusterID: cluster.ID.Uint64(),
		TCUNumber: tcuNumber,
	}

	created := false

	billingCtx := c.billingModel.User(c.tenantAndUser)
	pocEnabled, err := billingCtx.IsPOCEnabled()

	if err != nil {
		return err
	}

	err = db.Transaction(func(tx *gorm.DB) error {
		cnt, err := entity.NewCDCTCUMeteringRecordQuerySet(tx).CostIDEq(costID).Count()
		if err != nil {
			return errors.Annotate(err, "failed query tcu metering record")
		}
		if cnt > 0 {
			created = true
			return nil
		}
		return record.Create(tx)
	})

	if err != nil {
		return err
	}

	if created {
		return nil
	}

	labels := map[string]string{
		"tenantID":  fmt.Sprintf("%d", cluster.TenantID),
		"projectID": fmt.Sprintf("%d", cluster.ProjectID),
		"clusterID": fmt.Sprintf("%d", cluster.ID),
		"project":   cluster.Project.Name,
		"cluster":   cluster.Name,
	}

	if pocEnabled {
		labels["poc"] = "true"
	}

	region := cluster.ComponentsV2[0].Region.Name

	unit := fmt.Sprintf(TCUCostUnitPattern, cluster.Provider.Name, region)

	costs := billing2.UnitCostsInput{
		CostID:        costID,
		CostTimeStamp: costTime.Unix(),
		Unit:          unit,
		Cost:          int(tcuNumber) * int(HourInMinutes),
		Labels:        labels,
	}

	status := entity.CDCMeteringStatusDone

	err = billingCtx.MeasureCosts(&costs)
	if err != nil {
		c.Log().Error("failed send measure costs to billing2", zap.Error(err))
		status = entity.CDCMeteringStatusFailed
	}

	record.Status = status

	err = db.Transaction(func(tx *gorm.DB) error {
		return record.Update(tx, entity.CDCTCUMeteringRecordDBSchema.Status)
	})
	return err
}

func (c BgContext) SendTicdcMesaures(ctx context.Context) error {
	if !config.Global().BillingConfig.EnableCDCTCUMetering {
		return nil
	}

	var (
		err        error
		tenantID   entity.ID
		clusters   []entity.Cluster
		cdcRecords []entity.CDCTask
	)
	tenantID = c.tenantAndUser.TenantID
	clusters, err = c.clusterdao.FindClusters(dbutil.ClusterFilter{
		TenantID: &tenantID,
		Free:     pointer.BoolPtr(false),
		Statuss: []entity.ClusterStatus{
			entity.ClusterStatusNormal,
			entity.ClusterStatusScaling,
			entity.ClusterStatusDeleted},
	})
	if err != nil {
		return errors.Annotatef(err, "failed to fetch clusters records")
	}

	for _, cluster := range clusters {
		cdcRecords, err = c.dao.FindCDCTasks(cdcDao.CdcFilter{Deleted: pointer.BoolPtr(false), ClusterID: cluster.ID.Uint64(),
			Statuss: []entity.CDCStatus{entity.CDCStatusNormal, entity.CDCStatusFailed, entity.CDCStatusPause}})
		if err != nil {
			c.log.Error("failed to get cdc tasks list", zap.Error(err))
			continue
		}
		if len(cdcRecords) == 0 {
			c.Log().Info("cdc records is zero", zap.Uint64("cluster ID", cluster.ID.Uint64()))
			// delete cdc machine and k8s resources if no cdc task exists
			continue
		}

		err = c.createRecordAndSendMetering(cluster)
		if err != nil {
			c.Log().Error("failed to send ticdc metering", zap.Uint64("cluster ID", cluster.ID.Uint64()), zap.Error(err))
			continue
		}
	}
	return nil
}