package service

import (
	"context"
	"fmt"
	"time"

	infrav1 "github.com/tidbcloud/infra-api/api/v1"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	clusterconsts "github.com/tidbcloud/infra-provider/pkg/consts"
	cloudv1 "github.com/tidbcloud/ticdc-service/api/v1"
	"github.com/tidbcloud/ticdc-service/pkg/cloud"
	"github.com/tidbcloud/ticdc-service/pkg/clusterproxy"
	"github.com/tidbcloud/ticdc-service/pkg/config"
	"github.com/tidbcloud/ticdc-service/pkg/consts"
	"github.com/tidbcloud/ticdc-service/pkg/regionalservice"
	"github.com/tidbcloud/ticdc-service/pkg/tidbcluster"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/emptypb"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

type cdcServer struct {
	apiv1.UnimplementedTiCDCClusterServiceServer

	kubeClient  client.Client
	cfg         *config.Config
	proxyModel  clusterproxy.Model
	regionalCli regionalservice.Client
}

func NewTiCDCClusterServiceServer(cli client.Client,
	cfg *config.Config,
	proxyModel clusterproxy.Model,
	regionalCli regionalservice.Client) apiv1.TiCDCClusterServiceServer {
	s := &cdcServer{
		kubeClient:  cli,
		cfg:         cfg,
		proxyModel:  proxyModel,
		regionalCli: regionalCli,
	}
	return s
}

// CreateTiCDCCluster saves the TiCDCCluster to k8s CR
func (c cdcServer) CreateTiCDCCluster(ctx context.Context, req *apiv1.CreateTiCDCClusterRequest) (*apiv1.TiCDCCluster, error) {
	tenantId := req.TicdcCluster.TenantId
	projectId := req.TicdcCluster.ProjectId
	clusterId := req.TicdcCluster.ClusterId
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcCluster.ClusterId))
	cluster, err := c.regionalCli.GetCluster(ctx, req.TicdcCluster.ClusterId)
	if err != nil || cluster == nil {
		logger.Error("Failed to get cluster", zap.Any("cluster", cluster), zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", clusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", clusterId, err)
	}
	appbox, err := c.regionalCli.GetAPPBoxByTenantIDAndClusterID(ctx, tenantId, clusterId)
	if err != nil {
		logger.Error("Failed to get appbox for cluster", zap.String("clusterId", clusterId), zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", clusterId, err)
	}

	cdc := cloudv1.TiCDC{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.TicdcClusterId,
			Namespace: c.cfg.Namespace,
			Labels: buildLabelsForCreatingTiCDC(c.cfg.Provider, c.cfg.Region,
				tenantId, projectId, clusterId),
		},
		Spec: cloudv1.TiCDCSpec{
			ClusterID:    clusterId,
			Replicas:     int(req.TicdcCluster.NodeCount),
			BaseProxyURL: cluster.ProxyBaseUrl,
			// we can add UpdateTiCDCNodeVersion API to support k8s node upgrade
			NodeVersion:      appbox.Kubernetes.Version,
			InstanceType:     req.TicdcCluster.NodeInstanceType,
			MountCertificate: req.TicdcCluster.MountCertificate,
			Storage: cloudv1.StorageSpec{
				Type:          req.TicdcCluster.Storage.Type,
				SizeGi:        req.TicdcCluster.Storage.SizeGi,
				Iops:          req.TicdcCluster.Storage.Iops,
				ThroughputMbs: req.TicdcCluster.Storage.ThroughputMbs,
			},
			Configuration: &cloudv1.TiCDCConfiguration{
				BaseImage: req.TicdcCluster.Configuration.BaseImage,
				Tag:       req.TicdcCluster.Configuration.Tag,
				Config:    req.TicdcCluster.Configuration.Config,
			},
		},
	}

	// Save the TiCDCCluster object to Kubernetes
	if err := c.kubeClient.Create(ctx, &cdc); err != nil {
		if apierrors.IsAlreadyExists(err) {
			return nil, status.Errorf(codes.AlreadyExists, "TiCDC cluster %s already exists", clusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to create TiCDC cluster %s: %s", clusterId, err)
	}
	// Build the response object
	return c.buildTiCDCClusterResponse(ctx, cdc, true)
}

func (c cdcServer) ListTiCDCClusters(ctx context.Context, req *apiv1.ListTiCDCClustersRequest) (*apiv1.ListTiCDCClustersResponse, error) {
	// Create a list to hold the TiCDC clusters
	list := &cloudv1.TiCDCList{}

	// Define list options to filter by namespace
	listOptions := []client.ListOption{
		client.InNamespace(c.cfg.Namespace),
		client.MatchingLabelsSelector{
			Selector: buildLabelSetForQuery(req.ProjectId, req.TenantId, req.TidbClusterId).AsSelector(),
		},
	}

	// List the TiCDC clusters from Kubernetes
	if err := c.kubeClient.List(ctx, list, listOptions...); err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to list TiCDC clusters: %s", err)
	}

	// Build the response
	clusters := make([]*apiv1.TiCDCCluster, 0, len(list.Items))
	for _, item := range list.Items {
		if len(req.TicdcClusterId) > 0 &&
			!matchTiCDCClusterID(req.TicdcClusterId, item.Name) {
			continue
		}
		resp, err := c.buildTiCDCClusterResponse(ctx, item, true)
		if err != nil {
			return nil, status.Errorf(codes.Internal, "Failed to build TiCDC cluster response: %s", err)
		}
		clusters = append(clusters, resp)
	}
	return &apiv1.ListTiCDCClustersResponse{TicdcClusters: clusters}, nil
}

func (c cdcServer) GetTiCDCCluster(ctx context.Context, req *apiv1.GetTiCDCClusterRequest) (*apiv1.TiCDCCluster, error) {
	cdc, err := c.getTiCDCByID(ctx, req.TicdcClusterId)
	if err != nil {
		return nil, err
	}
	return c.buildTiCDCClusterResponse(ctx, *cdc, false)
}

func (c cdcServer) ScaleTiCDCCluster(ctx context.Context, req *apiv1.ScaleTiCDCClusterRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	cdc, err := c.getTiCDCByID(ctx, req.TicdcClusterId)
	if err != nil {
		return nil, err
	}

	// Update the number of replicas
	cdc.Spec.Replicas = int(req.Count)
	cdc.Spec.InstanceType = req.InstanceType
	cdc.Spec.Storage = cloudv1.StorageSpec{
		Type:          req.Storage.Type,
		SizeGi:        req.Storage.SizeGi,
		Iops:          req.Storage.Iops,
		ThroughputMbs: req.Storage.ThroughputMbs,
	}

	// Update the TiCDC cluster in Kubernetes
	if err := c.kubeClient.Update(ctx, cdc); err != nil {
		logger.Error("Failed to scale TiCDC cluster", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to scale TiCDC cluster %s: %s", req.TicdcClusterId, err)
	}

	return &emptypb.Empty{}, nil
}

func (c cdcServer) UpdateTiCDCClusterConfig(ctx context.Context, req *apiv1.UpdateTiCDCClusterConfigRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	cdc, err := c.getTiCDCByID(ctx, req.TicdcClusterId)
	if err != nil {
		return nil, err
	}
	// Update the number of replicas
	cdc.Spec.Configuration = &cloudv1.TiCDCConfiguration{
		BaseImage: req.Configuration.BaseImage,
		Tag:       req.Configuration.Tag,
		Config:    req.Configuration.Config,
	}

	// Update the TiCDC cluster in Kubernetes
	if err := c.kubeClient.Update(ctx, cdc); err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to scale TiCDC cluster %s: %s", req.TicdcClusterId, err)
	}
	logger.Info("TiCDC cluster configuration updated")
	return &emptypb.Empty{}, nil
}

func (c cdcServer) UpdateTiCDCClusterStorage(ctx context.Context, req *apiv1.UpdateTiCDCClusterStorageRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	cdc, err := c.getTiCDCByID(ctx, req.TicdcClusterId)
	if err != nil {
		return nil, err
	}
	// Update the storage
	cdc.Spec.Storage = cloudv1.StorageSpec{
		Type:          req.Storage.Type,
		SizeGi:        req.Storage.SizeGi,
		Iops:          req.Storage.Iops,
		ThroughputMbs: req.Storage.ThroughputMbs,
	}

	// Update the TiCDC cluster in Kubernetes
	if err := c.kubeClient.Update(ctx, cdc); err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to update TiCDC cluster storage %s: %s", req.TicdcClusterId, err)
	}
	logger.Info("TiCDC cluster storage updated")
	return &emptypb.Empty{}, nil
}

func (c cdcServer) DeleteTiCDCCluster(ctx context.Context, request *apiv1.DeleteTiCDCClusterRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", request.TicdcClusterId))
	cdc, err := c.getTiCDCByID(ctx, request.TicdcClusterId)
	if err != nil {
		logger.Error("Failed to get cdc cluster", zap.Error(err))
		return nil, err
	}
	if err := c.kubeClient.Delete(ctx, cdc); err != nil {
		if apierrors.IsNotFound(err) {
			logger.Warn("ticdc cluster not found")
			return &emptypb.Empty{}, nil
		}
		logger.Error("Failed to delete cdc cluster", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to delete cdc cluster %s: %s", request.TicdcClusterId, err)
	}
	return &emptypb.Empty{}, nil
}

func (c cdcServer) RestartTiCDCCluster(ctx context.Context, req *apiv1.RestartTiCDCClusterRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	cdc, err := c.getTiCDCByID(ctx, req.TicdcClusterId)
	if err != nil {
		logger.Error("Failed to get cdc cluster", zap.Error(err))
		return nil, err
	}

	if len(cdc.ObjectMeta.Annotations) == 0 {
		cdc.ObjectMeta.Annotations = make(map[string]string)
	}
	cdc.ObjectMeta.Annotations[consts.RestartTiCDCAnnotation] = time.Now().UTC().Format(time.RFC3339)
	if err := c.kubeClient.Update(ctx, cdc); err != nil {
		logger.Error("Failed to restart TiCDC cluster", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to restart TiCDC cluster %s: %s", req.TicdcClusterId, err)
	}

	return &emptypb.Empty{}, nil
}

func (c cdcServer) UpdateTiCDCNodeVersion(ctx context.Context, req *apiv1.UpdateTiCDCNodeVersionRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	cdc, err := c.getTiCDCByID(ctx, req.TicdcClusterId)
	if err != nil {
		logger.Error("Failed to get cdc cluster", zap.Error(err))
		return nil, err
	}

	if cdc.Spec.NodeVersion == req.NodeVersion {
		logger.Info("TiCDC cluster node version already matches, no update needed", zap.String("nodeVersion", req.NodeVersion))
		return &emptypb.Empty{}, nil
	}

	cdc.Spec.NodeVersion = req.NodeVersion
	if err := c.kubeClient.Update(ctx, cdc); err != nil {
		logger.Error("Failed to update TiCDC cluster node version", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to update TiCDC cluster %s node version: %s", req.TicdcClusterId, err)
	}
	logger.Info("TiCDC cluster node version updated", zap.String("nodeVersion", req.NodeVersion))
	return &emptypb.Empty{}, nil
}

func (c cdcServer) MountCertificateTiCDCCluster(ctx context.Context, req *apiv1.MountCertificateTiCDCClusterRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	cdc, err := c.getTiCDCByID(ctx, req.TicdcClusterId)
	if err != nil {
		logger.Error("Failed to get cdc cluster", zap.Error(err))
		return nil, err
	}

	cdc.Spec.MountCertificate = true
	if err := c.kubeClient.Update(ctx, cdc); err != nil {
		logger.Error("Failed to mount certificate for TiCDC cluster", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to mount certificate for TiCDC cluster %s: %s", req.TicdcClusterId, err)
	}
	return &emptypb.Empty{}, nil
}

func (c cdcServer) GetTiCDCClusterRole(ctx context.Context, req *apiv1.GetTiCDCClusterRoleRequest) (*apiv1.TiCDCClusterRole, error) {
	clusterNS := fmt.Sprintf("%s-%s", c.cfg.ClusterNSPrefix, req.ProjectId)
	tidbClusterProvider, err := tidbcluster.NewClusterCRProvider(ctx, c.kubeClient, clusterNS, req.TidbClusterId, c.cfg.Region)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to create tidbcluster provider for TiDB cluster %s: %s", req.TidbClusterId, err.Error())
	}
	tidbClusterRole, err := tidbClusterProvider.GetClusterRole(ctx, req.TidbClusterId)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get the role of tidb cluster %s: %s", req.TidbClusterId, err.Error())
	}
	appbox, err := c.regionalCli.GetAPPBoxByTenantIDAndProjectID(ctx, req.TenantId, req.ProjectId)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get appbox by tenant id %s and project id %s: %s", req.TenantId, req.ProjectId, err.Error())
	}

	role := &apiv1.TiCDCClusterRole{
		CloudIamIdentity: tidbClusterRole.CloudIamIdentity,
		CloudAccountId:   appbox.Kubernetes.AccountId,
		AwsExternalId:    tidbClusterRole.ExternalId,
	}
	if c.cfg.Provider == config.ProviderAWS {
		role.CloudIamIdentity = appbox.Kubernetes.GetAws().TicdcNodeRole
	}
	return role, nil
}

func (c cdcServer) CreateChangefeed(ctx context.Context, req *apiv1.CreateChangefeedRequest) (*apiv1.Changefeed, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	var (
		cf  *apiv1.Changefeed
		err error
	)
	if err = c.doInProxy(ctx, req.TicdcClusterId, func(proxy clusterproxy.ClusterProxy) error {
		req.Changefeed.ChangefeedId = req.ChangefeedId
		cf, err = proxy.ChangefeedClient().CreateChangefeedV2(ctx, req.TicdcClusterId, req.Changefeed)
		return err
	}); err != nil {
		logger.Error("Failed to create changefeed", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to create changefeed: %s", err)
	}
	return cf, nil
}

func (c cdcServer) ListChangefeeds(ctx context.Context, request *apiv1.ListChangefeedsRequest) (*apiv1.ListChangefeedsResponse, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", request.TicdcClusterId))
	var (
		cfList *apiv1.ListChangefeedsResponse
		err    error
	)
	if err = c.doInProxy(ctx, request.TicdcClusterId, func(proxy clusterproxy.ClusterProxy) error {
		cfList, err = proxy.ChangefeedClient().GetChangefeedListV2(ctx, request.TicdcClusterId)
		return err
	}); err != nil {
		logger.Error("Failed to list changefeed", zap.Error(err))
		return nil, err
	}
	return cfList, nil
}

func (c cdcServer) GetChangefeed(ctx context.Context, request *apiv1.GetChangefeedRequest) (*apiv1.Changefeed, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", request.TicdcClusterId))
	var (
		cf  *apiv1.Changefeed
		err error
	)
	if err := c.doInProxy(ctx, request.TicdcClusterId, func(proxy clusterproxy.ClusterProxy) error {
		cf, err = proxy.ChangefeedClient().GetChangefeedV2(ctx, request.TicdcClusterId, request.ChangefeedId)
		return err
	}); err != nil {
		logger.Error("Failed to get changefeed", zap.Error(err))
		return nil, err
	}
	return cf, err
}

func (c cdcServer) PauseChangefeed(ctx context.Context, req *apiv1.PauseChangefeedRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	if err := c.doInProxy(ctx, req.TicdcClusterId, func(proxy clusterproxy.ClusterProxy) error {
		return proxy.ChangefeedClient().PauseChangefeedV2(ctx, req.TicdcClusterId, req.ChangefeedId)
	}); err != nil {
		logger.Error("Failed to pause changefeed", zap.Error(err))
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (c cdcServer) ResumeChangefeed(ctx context.Context, req *apiv1.ResumeChangefeedRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	if err := c.doInProxy(ctx, req.TicdcClusterId, func(proxy clusterproxy.ClusterProxy) error {
		return proxy.ChangefeedClient().ResumeChangefeedV2(ctx, req.TicdcClusterId, req.ChangefeedId, req.Config)
	}); err != nil {
		logger.Error("Failed to pause changefeed", zap.Error(err))
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (c cdcServer) UpdateChangefeed(ctx context.Context, request *apiv1.UpdateChangefeedRequest) (*apiv1.Changefeed, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", request.TicdcClusterId))
	var (
		retCf *apiv1.Changefeed
		err   error
	)
	if err = c.doInProxy(ctx, request.TicdcClusterId, func(proxy clusterproxy.ClusterProxy) error {
		retCf, err = proxy.ChangefeedClient().UpdateChangefeedV2(ctx,
			request.TicdcClusterId, request.ChangefeedId, request.Changefeed)
		return err
	}); err != nil {
		logger.Error("Failed to create changefeed", zap.Error(err))
		return nil, err
	}
	return retCf, nil
}

func (c cdcServer) DeleteChangefeed(ctx context.Context, req *apiv1.DeleteChangefeedRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	if err := c.doInProxy(ctx, req.TicdcClusterId, func(proxy clusterproxy.ClusterProxy) error {
		return proxy.ChangefeedClient().DeleteChangefeedV2(ctx, req.TicdcClusterId, req.ChangefeedId)
	}); err != nil {
		logger.Error("Failed to delete changefeed", zap.Error(err))
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (c cdcServer) CompensateComputeCloudResourceTags(ctx context.Context, req *apiv1.CompensateComputeCloudResourceTagsRequest) (*emptypb.Empty, error) {
	if c.cfg.Provider != config.ProviderAWS {
		return &emptypb.Empty{}, nil
	}

	compensater := cloud.NewAwsResourceTagCompensater(c.cfg)
	if err := compensater.CompensateComputeTags(ctx, c.kubeClient); err != nil {
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (c cdcServer) getTiCDCByID(ctx context.Context, id string) (*cloudv1.TiCDC, error) {
	// Create a TiCDC object to hold the retrieved data
	cdc := &cloudv1.TiCDC{}
	// Define the key to identify the TiCDC cluster
	key := client.ObjectKey{
		Namespace: c.cfg.Namespace,
		Name:      id,
	}
	// Get the TiCDC cluster from Kubernetes
	if err := c.kubeClient.Get(ctx, key, cdc); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "TiCDC cluster %s not found", id)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get TiCDC cluster %s: %s", id, err)
	}
	return cdc, nil
}

// ticdc operator version is the same as tidb operator version
func (c cdcServer) getTiCDCOperatorVersion(ctx context.Context, clusterId string, projectID string) (string, error) {
	clusterNS := fmt.Sprintf("%s-%s", c.cfg.ClusterNSPrefix, projectID)
	tidbclusterProvider, err := tidbcluster.NewClusterCRProvider(ctx, c.kubeClient, clusterNS, clusterId, c.cfg.Region)
	if err != nil {
		return "", status.Errorf(codes.Internal, "Failed to create tidbcluster provider for TiDB cluster %s: %s", clusterId, err.Error())
	}
	v2Enabled, err := tidbclusterProvider.FeatureEnabled(
		ctx,
		clusterId,
		clusterconsts.ClusterFeatureUseTiDBOperatorV2,
	)
	if err != nil {
		return "", status.Errorf(codes.Internal, "Failed to check if TiDB cluster %s is using TiDB Operator v2: %s", clusterId, err.Error())
	}
	if v2Enabled {
		return "v2", nil
	}
	return "v1", nil
}

func (c cdcServer) doInProxy(
	ctx context.Context,
	cdcClusterID string,
	f func(proxy clusterproxy.ClusterProxy) error) error {
	logger := tilog.WithContextL(ctx)
	cdc, err := c.getTiCDCByID(ctx, cdcClusterID)
	if err != nil {
		logger.Error("Failed to get cdc cluster", zap.Error(err))
		return err
	}
	proxy, err := c.proxyModel.GetClusterProxy(ctx, cdc.Labels[infrav1.ClusterTagKey], clusterproxy.CDCHeavyOperationTimeout)
	if err != nil {
		logger.Error("Failed to get proxy", zap.Error(err))
		return err
	}
	return f(proxy)
}

func (c cdcServer) buildTiCDCClusterResponse(ctx context.Context, item cloudv1.TiCDC, ignoreOperatorVersion bool) (*apiv1.TiCDCCluster, error) {
	cluster := &apiv1.TiCDCCluster{
		Id:               item.Name,
		ProjectId:        item.Labels[infrav1.ProjectTagKey],
		TenantId:         item.Labels[infrav1.TenantTagKey],
		ClusterId:        item.Labels[infrav1.ClusterTagKey],
		NodeInstanceType: item.Spec.InstanceType,
		NodeCount:        uint32(item.Spec.Replicas),
		MountCertificate: item.Spec.MountCertificate,
		NodeVersion:      item.Spec.NodeVersion,
		Storage: &apiv1.TiCDCCluster_Storage{
			Type:          item.Spec.Storage.Type,
			SizeGi:        item.Spec.Storage.SizeGi,
			Iops:          item.Spec.Storage.Iops,
			ThroughputMbs: item.Spec.Storage.ThroughputMbs,
		},
		Configuration: &apiv1.TiCDCCluster_Configuration{
			BaseImage: item.Spec.Configuration.BaseImage,
			Tag:       item.Spec.Configuration.Tag,
			Config:    item.Spec.Configuration.Config,
		},
		Phase: mapTiCDCClusterPhase(item.Status.Phase),
	}
	if item.Status.NodeInfo != nil {
		cluster.LastAppliedNodeSpec = &apiv1.TiCDCCluster_NodeSpec{
			InstanceType: item.Status.NodeInfo.InstanceType,
			Count:        uint32(item.Status.NodeInfo.Replicas),
		}
	}
	if cluster.MountCertificate {
		cluster.CertificateBaseDir = consts.TiCDCCertMountPath
	}

	if ignoreOperatorVersion {
		return cluster, nil
	}

	operatorVersion, err := c.getTiCDCOperatorVersion(ctx, item.Spec.ClusterID, cluster.ProjectId)
	if err != nil {
		return nil, fmt.Errorf("failed to get operator version: %w", err)
	}
	cluster.OperatorVersion = operatorVersion
	return cluster, nil
}

func matchTiCDCClusterID(filters []string, ticdcClusterID string) bool {
	for _, f := range filters {
		if f == ticdcClusterID {
			return true
		}
	}
	return false
}

var cdcClusterPhaseMap = map[cloudv1.TiCDCPhase]apiv1.TiCDCCluster_Phase{
	cloudv1.TiCDCPhaseCreating:    apiv1.TiCDCCluster_CREATING,
	cloudv1.TiCDCPhaseRunning:     apiv1.TiCDCCluster_RUNNING,
	cloudv1.TiCDCPhaseUnavailable: apiv1.TiCDCCluster_UNAVAILABLE,
	cloudv1.TiCDCPhaseScaling:     apiv1.TiCDCCluster_SCALING,
	cloudv1.TiCDCPhaseUpgrading:   apiv1.TiCDCCluster_UPGRADING,
	cloudv1.TiCDCPhaseDeleting:    apiv1.TiCDCCluster_DELETING,
}

func mapTiCDCClusterPhase(phase cloudv1.TiCDCPhase) apiv1.TiCDCCluster_Phase {
	if v, ok := cdcClusterPhaseMap[phase]; ok {
		return v
	}
	return apiv1.TiCDCCluster_PHASE_UNSPECIFIED
}