package changefeed

import (
	"context"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/tidbcloud/dataflow-service/app/models/biz_changefeed"
	"github.com/tidbcloud/dataflow-service/app/models/config_manager"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/ticdc_spec"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	changefeedopenbp "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/openapi/v1beta1"
	ticdc_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

const defaultListChangefeedsPageSize uint32 = 20

var _ ChangefeedAdapter = (*connectorChangefeedAdapter)(nil)

type changefeedScope struct {
	Cluster *mgmt.Cluster
}

func newChangefeedScope(cluster *mgmt.Cluster) changefeedScope {
	return changefeedScope{Cluster: cluster}
}

func (s changefeedScope) tenantID() uint64 {
	if s.Cluster == nil {
		return 0
	}
	return s.Cluster.GetOrganizationId()
}

func (s changefeedScope) projectID() uint64 {
	if s.Cluster == nil {
		return 0
	}
	return s.Cluster.GetProjectId()
}

func (s changefeedScope) clusterID() uint64 {
	if s.Cluster == nil {
		return 0
	}
	return s.Cluster.GetClusterId()
}

type connectorChangefeedAdapter struct {
	connectorAPI            ConnectorServiceServer
	instanceTypeModel       ticdc_spec.CDCInstanceTypeModel
	changefeedModel         biz_changefeed.BizChangefeedModel
	bizChangefeedRepo       repo.BizChangefeedRepo
	ticdcClusterRepo        repo.TiCDCClusterRepo
	cdcClusterConfigManager config_manager.TiCDCClusterConfigManager
}

func newConnectorChangefeedAdapter(
	connectorAPI ConnectorServiceServer,
	instanceTypeModel ticdc_spec.CDCInstanceTypeModel,
	changefeedModel biz_changefeed.BizChangefeedModel,
	bizChangefeedRepo repo.BizChangefeedRepo,
	ticdcClusterRepo repo.TiCDCClusterRepo,
	cdcClusterConfigManager config_manager.TiCDCClusterConfigManager,
) ChangefeedAdapter {
	return &connectorChangefeedAdapter{
		connectorAPI:            connectorAPI,
		instanceTypeModel:       instanceTypeModel,
		changefeedModel:         changefeedModel,
		bizChangefeedRepo:       bizChangefeedRepo,
		ticdcClusterRepo:        ticdcClusterRepo,
		cdcClusterConfigManager: cdcClusterConfigManager,
	}
}

func parseUintID(raw, field string) (uint64, error) {
	v, err := strconv.ParseUint(raw, 10, 64)
	if err != nil {
		return 0, bizerr.New(bizerr.ErrCodeParams, "%s is invalid", field)
	}
	return v, nil
}

func (a *connectorChangefeedAdapter) ListChangefeeds(ctx context.Context,
	req *changefeedopenbp.ListChangefeedsRequest,
	scope changefeedScope,
) (*changefeedopenbp.ListChangefeedsResponse, error) {
	pageSize := normalizeListPageSize(req.GetPageSize())
	page, err := decodePageToken(req.GetPageToken())
	if err != nil {
		return nil, err
	}

	connectorType, err := connectorTypeFromDownstreamType(req.GetDownstreamType(), false)
	if err != nil {
		return nil, err
	}

	connectors, err := a.connectorAPI.ListConnectors(ctx, &dataflow.ConnectorListReq{
		ClusterId:     scope.clusterID(),
		ProjectId:     scope.projectID(),
		OrgId:         scope.tenantID(),
		Page:          page,
		PerPage:       pageSize,
		ConnectorType: connectorType,
	})
	if err != nil {
		return nil, err
	}
	if connectors == nil {
		connectors = &dataflow.Connectors{}
	}

	out := &changefeedopenbp.ListChangefeedsResponse{
		Changefeeds: make([]*changefeedopenbp.Changefeed, 0, len(connectors.GetItems())),
		TotalSize:   int32(connectors.GetTotal()),
	}
	for _, connector := range connectors.GetItems() {
		out.Changefeeds = append(out.Changefeeds, convertConnectorToChangefeed(connector))
	}

	if hasNextPage(page, pageSize, connectors.GetTotal()) {
		token, err := encodePageToken(page + 1)
		if err != nil {
			return nil, err
		}
		out.NextPageToken = token
	}

	return out, nil
}

func (a *connectorChangefeedAdapter) ResolveChangefeedScope(ctx context.Context,
	changefeedID string,
) (changefeedScope, error) {
	connector, err := a.connectorByID(ctx, changefeedID)
	if err != nil {
		return changefeedScope{}, err
	}
	req := connector.GetRequest()
	if req == nil {
		return changefeedScope{}, bizerr.New(bizerr.ErrCodeParams, "changefeed request is missing")
	}
	return newChangefeedScope(&mgmt.Cluster{
		OrganizationId: req.GetOrgId(),
		ProjectId:      req.GetProjectId(),
		ClusterId:      req.GetClusterId(),
	}), nil
}

func (a *connectorChangefeedAdapter) CreateChangefeed(ctx context.Context,
	req *changefeedopenbp.CreateChangefeedRequest,
	scope changefeedScope,
) (*changefeedopenbp.Changefeed, error) {
	createReq, err := a.toConnectorCreateReq(req, scope)
	if err != nil {
		return nil, err
	}

	createdID, err := a.connectorAPI.CreateConnector(ctx, createReq)
	if err != nil {
		return nil, err
	}
	if req.DryRun {
		return &changefeedopenbp.Changefeed{}, nil
	}

	connector, err := a.connectorAPI.GetConnector(ctx, createdID)
	if err != nil {
		return nil, err
	}
	return convertConnectorToChangefeed(connector), nil
}

func (a *connectorChangefeedAdapter) DeleteChangefeed(ctx context.Context,
	req *changefeedopenbp.DeleteChangefeedRequest,
	scope changefeedScope,
) (*emptypb.Empty, error) {
	changefeedID, err := parseUintID(req.GetChangefeedId(), "changefeed_id")
	if err != nil {
		return nil, err
	}
	_, err = a.connectorAPI.DeleteConnector(ctx, connectorIDWithScope(changefeedID, scope))
	if err != nil {
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (a *connectorChangefeedAdapter) EditChangefeedDownstreamConfig(ctx context.Context,
	req *changefeedopenbp.EditChangefeedDownstreamConfigRequest,
	scope changefeedScope,
) (*changefeedopenbp.Changefeed, error) {
	connector, err := a.connectorByID(ctx, req.GetChangefeedId())
	if err != nil {
		return nil, err
	}

	editReq, err := a.connectorEditReqFromCreateReq(connector, req)
	if err != nil {
		return nil, err
	}
	if _, err := a.connectorAPI.EditConnector(ctx, editReq); err != nil {
		return nil, err
	}
	if req.DryRun {
		return &changefeedopenbp.Changefeed{}, nil
	}

	updated, err := a.connectorAPI.GetConnector(ctx, connectorIDWithScope(connector.GetConnectorId(), scope))
	if err != nil {
		return nil, err
	}
	return convertConnectorToChangefeed(updated), nil
}

func (a *connectorChangefeedAdapter) GetChangefeed(ctx context.Context,
	req *changefeedopenbp.GetChangefeedRequest,
	_ changefeedScope,
) (*changefeedopenbp.Changefeed, error) {
	connector, err := a.connectorByID(ctx, req.GetChangefeedId())
	if err != nil {
		return nil, err
	}
	if req.GetInvalidateCache() {
		if err := a.refreshRealtimeCheckpoint(ctx, connector); err != nil {
			return nil, err
		}
	}
	return convertConnectorToChangefeed(connector), nil
}

// GetChangefeedKernelInfo returns the TiCDC kernel info (version, image tag and
// kernel type) of the changefeed. It serves the internal-only
// GetChangefeedKernelInfo OpenAPI. The changefeed's authority was already
// established by the service (changefeed scope + RBAC); here we follow the
// changefeed's BizChangefeed binding straight to its TiCDC cluster.
func (a *connectorChangefeedAdapter) GetChangefeedKernelInfo(ctx context.Context,
	req *changefeedopenbp.GetChangefeedKernelInfoRequest,
) (*changefeedopenbp.ChangefeedKernelInfo, error) {
	changefeedID, err := parseUintID(req.GetChangefeedId(), "changefeed_id")
	if err != nil {
		return nil, err
	}

	bizChangefeed, bizErr := a.bizChangefeedRepo.Get(ctx, func(qs entity.BizChangefeedQuerySet) entity.BizChangefeedQuerySet {
		return qs.IDIn(entity.ID(changefeedID))
	})
	if bizErr != nil {
		return nil, bizErr.Chain("get changefeed %d binding", changefeedID)
	}
	ticdcClusterID := uint64(bizChangefeed.TiCDCClusterID)
	if ticdcClusterID == 0 {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "changefeed %d has no bound ticdc cluster", changefeedID)
	}

	ticdcCluster, bizErr := a.ticdcClusterRepo.Load(ctx, ticdcClusterID)
	if bizErr != nil {
		return nil, bizErr.Chain("load ticdc cluster %d", ticdcClusterID)
	}

	versionTag, bizErr := a.cdcClusterConfigManager.GetVersionAndImageTag(ctx,
		&domain.GetTiCDCClusterVersionAndImageTagReq{TiCDCClusterID: ticdcClusterID})
	if bizErr != nil {
		return nil, bizErr.Chain("get ticdc cluster %d component version and image tag", ticdcClusterID)
	}

	kernelType := changefeedopenbp.ChangefeedKernelInfo_KERNEL_TYPE_CLASSIC
	if ticdcCluster.IsNewKernel {
		kernelType = changefeedopenbp.ChangefeedKernelInfo_KERNEL_TYPE_NEW
	}
	return &changefeedopenbp.ChangefeedKernelInfo{
		ChangefeedId: req.GetChangefeedId(),
		TicdcVersion: versionTag.Version,
		ImageTag:     versionTag.ImageTag,
		KernelType:   kernelType,
	}, nil
}

func (a *connectorChangefeedAdapter) PauseChangefeed(ctx context.Context,
	req *changefeedopenbp.PauseChangefeedRequest,
	scope changefeedScope,
) (*emptypb.Empty, error) {
	changefeedID, err := parseUintID(req.GetChangefeedId(), "changefeed_id")
	if err != nil {
		return nil, err
	}
	_, err = a.connectorAPI.StopConnector(ctx, connectorIDWithScope(changefeedID, scope))
	if err != nil {
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (a *connectorChangefeedAdapter) ResumeChangefeed(ctx context.Context,
	req *changefeedopenbp.ResumeChangefeedRequest,
	scope changefeedScope,
) (*emptypb.Empty, error) {
	changefeedID, err := parseUintID(req.GetChangefeedId(), "changefeed_id")
	if err != nil {
		return nil, err
	}
	_, err = a.connectorAPI.StartConnector(ctx, connectorIDWithScope(changefeedID, scope))
	if err != nil {
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (a *connectorChangefeedAdapter) ListChangefeedRCUs(ctx context.Context,
	req *changefeedopenbp.ListChangefeedRCUsRequest,
) (*changefeedopenbp.ListChangefeedRCUsResponse, error) {
	_ = req
	resp, err := a.instanceTypeModel.ListChangefeedInstanceTypes(ctx)
	if err != nil {
		return nil, err
	}
	rcus := make([]*changefeedopenbp.ChangefeedRCU, 0, len(resp.InstanceTypes))
	for _, item := range resp.InstanceTypes {
		rcus = append(rcus, &changefeedopenbp.ChangefeedRCU{
			Name:        item.Name,
			DisplayName: fmt.Sprintf("%d RCUs", item.RCUNum),
			RcuNum:      uint32(item.RCUNum),
			RpsLimit:    uint32(item.RPSLimit),
		})
	}
	return &changefeedopenbp.ListChangefeedRCUsResponse{Rcus: rcus}, nil
}

func (a *connectorChangefeedAdapter) ScaleChangefeed(ctx context.Context,
	req *changefeedopenbp.ScaleChangefeedRequest,
	scope changefeedScope,
) (*changefeedopenbp.Changefeed, error) {
	changefeedID, err := parseUintID(req.GetChangefeedId(), "changefeed_id")
	if err != nil {
		return nil, err
	}

	_, err = a.connectorAPI.ScaleChangefeed(ctx, &dataflow.ChangefeedScaleReq{
		ChangefeedId:           changefeedID,
		ClusterId:              scope.clusterID(),
		OrgId:                  scope.tenantID(),
		ProjectId:              scope.projectID(),
		ChangefeedInstanceType: req.GetReplicationCapacity(),
	})
	if err != nil {
		return nil, err
	}

	updated, err := a.connectorAPI.GetConnector(ctx, connectorIDWithScope(changefeedID, scope))
	if err != nil {
		return nil, err
	}
	return convertConnectorToChangefeed(updated), nil
}

func (a *connectorChangefeedAdapter) GetEventFilters(ctx context.Context,
	req *changefeedopenbp.GetEventFiltersRequest,
	_ changefeedScope,
) (*changefeedopenbp.EventFilters, error) {
	connector, err := a.connectorByID(ctx, req.GetChangefeedId())
	if err != nil {
		return nil, err
	}
	return convertConnectorToOpenAPIFilters(connector)
}

func (a *connectorChangefeedAdapter) SetEventFilters(ctx context.Context,
	req *changefeedopenbp.SetEventFiltersRequest,
	_ changefeedScope,
) (*changefeedopenbp.SetEventFiltersResponse, error) {
	connector, err := a.connectorByID(ctx, req.GetChangefeedId())
	if err != nil {
		return nil, err
	}

	editReq, err := newConnectorEditRequest(connector, req.GetEventFilters())
	if err != nil {
		return nil, err
	}

	if _, err := a.connectorAPI.EditConnector(ctx, editReq); err != nil {
		return nil, err
	}
	return &changefeedopenbp.SetEventFiltersResponse{EventFilters: req.GetEventFilters()}, nil
}

func (a *connectorChangefeedAdapter) connectorByID(ctx context.Context,
	changefeedID string,
) (*dataflow.Connector, error) {
	cfID, err := parseUintID(changefeedID, "changefeed_id")
	if err != nil {
		return nil, err
	}
	connector, err := a.connectorAPI.GetConnector(ctx, &dataflow.ConnectorID{ConnectorId: cfID})
	if err != nil {
		return nil, err
	}
	if connector == nil {
		return nil, bizerr.New(bizerr.ErrCodeGRPC, "changefeed not found")
	}
	return connector, nil
}

func (a *connectorChangefeedAdapter) refreshRealtimeCheckpoint(
	ctx context.Context,
	connector *dataflow.Connector,
) error {
	logger := log.WithContextS(ctx).With(zap.Uint64("connector_id", connector.GetConnectorId()))

	detail, err := a.changefeedModel.GetChangefeedDetail(ctx, &domain.GetBizChangefeedDetailReq{
		BizChangefeedID: connector.GetConnectorId(),
	})
	if err != nil {
		if err.RawErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil
		}
		return err.Chain("get realtime changefeed detail")
	}

	kernelChangefeed, err := parseChangefeedDetail(logger, detail)
	if err != nil {
		return err.Chain("parse realtime changefeed detail")
	}

	// example data:
	// {"checkpoint_time": "2026-06-24 03:39:02.916", "checkpoint_ts": "467212001061371908", ...}
	if kernelChangefeed.CheckpointTs == 0 {
		logger.Warn("realtime changefeed checkpoint_ts is 0, skipping update")
		return nil
	}
	connector.CheckpointUtc = timestamppb.New(time.Unix(utils.TransToTS(kernelChangefeed.CheckpointTs), 0))
	connector.CheckpointTso = kernelChangefeed.CheckpointTs
	return nil
}

func parseChangefeedDetail(logger *zap.SugaredLogger, detail *domain.GetBizChangefeedDetailResp) (*ticdc_v1.Changefeed, bizerr.BizError) {
	if detail == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "changefeed detail is nil")
	}
	if detail.Changefeed == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "changefeed detail.changefeed is nil")
	}

	changefeedBytes, mErr := detail.Changefeed.MarshalJSON()
	if mErr != nil {
		return nil, bizerr.WrapError(mErr, "marshal changefeed config")
	}
	regionalChangefeed := new(ticdc_v1.Changefeed)
	if err := protojson.Unmarshal(changefeedBytes, regionalChangefeed); err != nil {
		return nil, bizerr.WrapError(err, "unmarshal changefeed config")
	}

	logger.Info("changefeed config from kernel",
		zap.String("state", regionalChangefeed.State),
		zap.Uint64("checkpointTs", regionalChangefeed.CheckpointTs),
		zap.String("checkpointTime", regionalChangefeed.CheckpointTime),
		zap.Uint64("resolvedTs", regionalChangefeed.ResolvedTs),
	)
	return regionalChangefeed, nil
}

func (a *connectorChangefeedAdapter) toConnectorCreateReq(oreq *changefeedopenbp.CreateChangefeedRequest,
	scope changefeedScope,
) (*dataflow.ConnectorCreateReq, error) {
	cf := oreq.GetChangefeed()
	if cf == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "changefeed is required")
	}

	connectorType, err := connectorTypeFromDownstreamType(cf.GetDownstreamType(), true)
	if err != nil {
		return nil, err
	}

	networkInfo, err := toDataflowNetworkInfo(cf.GetNetworkInfo(), scope.Cluster)
	if err != nil {
		return nil, err
	}

	req := &dataflow.ConnectorCreateReq{
		ClusterId:              scope.clusterID(),
		ProjectId:              scope.projectID(),
		OrgId:                  scope.tenantID(),
		ConnectorType:          connectorType,
		Name:                   cf.GetName(),
		ChangefeedInstanceType: cf.GetReplicationCapacity(),
		NetworkInfo:            networkInfo,
		Internal:               false,
		KafkaInfo:              nil, // filled later based on downstream type
		MysqlInfo:              nil, // filled later based on downstream type
		S3Info:                 nil, // filled later based on downstream type
		GcsInfo:                nil, // filled later based on downstream type
		ServerlessTidbInfo:     nil, // filled later based on downstream type
		PulsarInfo:             nil, // filled later based on downstream type
		AzureBlobStorageInfo:   nil, // filled later based on downstream type
		DryRun:                 oreq.DryRun,
	}

	switch connectorType {
	case dataflow.ConnectorType_KAFKA:
		req.KafkaInfo = toDataflowKafkaInfo(cf, scope)
		if req.KafkaInfo == nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "kafka configuration is required")
		}
		if err := validateKafkaCreateInput(req); err != nil {
			return nil, err
		}
	case dataflow.ConnectorType_MYSQL, dataflow.ConnectorType_TIDB, dataflow.ConnectorType_SERVERLESS_TIDB:
		req.MysqlInfo = toDataflowMySQLInfo(cf, scope)
		if req.MysqlInfo == nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "mysql configuration is required")
		}
	case dataflow.ConnectorType_S3:
		req.S3Info = toDataflowS3Info(cf, scope)
		if req.S3Info == nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "s3 configuration is required")
		}
	case dataflow.ConnectorType_GCS:
		req.GcsInfo = toDataflowGcsInfo(cf, scope)
		if req.GcsInfo == nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "gcs configuration is required")
		}
	case dataflow.ConnectorType_AZURE_BLOB_STORAGE:
		req.AzureBlobStorageInfo = toDataflowAzureBlobInfo(cf, scope)
		if req.AzureBlobStorageInfo == nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "azure blob storage configuration is required")
		}
	default:
		return nil, bizerr.New(bizerr.ErrCodeParams, "unsupported downstream type")
	}
	return req, nil
}

func requiresKafkaBrokerEndpoints(networkInfo *dataflow.NetworkInfo) bool {
	if networkInfo == nil {
		return true
	}
	switch networkInfo.GetNetworkType() {
	case dataflow.NetworkType_NetworkType_Private_Link,
		dataflow.NetworkType_NetworkType_Private_Service_Connect,
		dataflow.NetworkType_NetworkType_Manually_Private_Link,
		dataflow.NetworkType_NetworkType_Manually_Private_Service_Connect:
		return false
	default:
		return true
	}
}

func validateKafkaCreateInput(req *dataflow.ConnectorCreateReq) error {
	kafkaInfo := req.GetKafkaInfo()
	if kafkaInfo == nil || kafkaInfo.GetBroker() == nil {
		return bizerr.New(bizerr.ErrCodeParams, "kafka.broker is required")
	}

	if requiresKafkaBrokerEndpoints(req.GetNetworkInfo()) &&
		strings.TrimSpace(kafkaInfo.GetBroker().GetBrokerEndpoints()) == "" {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"kafka.broker.broker_endpoints is required when network_info.network_type is PUBLIC or VPC_PEERING",
		)
	}
	return nil
}

func (a *connectorChangefeedAdapter) mergeChangefeedEdit(connector *dataflow.Connector,
	req *changefeedopenbp.EditChangefeedDownstreamConfigRequest,
) (*dataflow.ConnectorCreateReq, error) {
	if connector == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "changefeed is required")
	}
	currentReq := connector.GetRequest()
	if currentReq == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "changefeed request is missing")
	}

	requestedType := req.GetDownstreamType()
	currentType := downstreamTypeFromConnectorType(connector.GetConnectorType())
	if requestedType != currentType {
		return nil, bizerr.New(bizerr.ErrCodeParams, "downstream_type cannot be changed")
	}

	merged := proto.Clone(currentReq).(*dataflow.ConnectorCreateReq)
	var err error
	switch connector.GetConnectorType() {
	case dataflow.ConnectorType_KAFKA:
		merged.KafkaInfo, err = a.mergeKafkaInfo(merged.GetKafkaInfo(), req)
	case dataflow.ConnectorType_MYSQL, dataflow.ConnectorType_TIDB:
		merged.MysqlInfo, err = a.mergeMySQLInfo(merged.GetMysqlInfo(), req)
	case dataflow.ConnectorType_SERVERLESS_TIDB:
		merged.MysqlInfo, err = a.mergeMySQLInfo(effectiveMySQLInfo(currentReq, connector.GetConnectorType()), req)
		// Guard against nil ServerlessTidbInfo to avoid panics when merging edits
		if currentReq.GetServerlessTidbInfo() == nil || connector.GetRequest().GetServerlessTidbInfo() == nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "serverless_tidb_info is missing for serverless downstream")
		}
		merged.ServerlessTidbInfo = &dataflow.ServerlessTidbInfo{
			ClusterId: connector.GetRequest().GetServerlessTidbInfo().GetClusterId(),
			MysqlInfo: merged.GetMysqlInfo(),
		}
		merged.MysqlInfo = nil
	case dataflow.ConnectorType_S3:
		merged.S3Info, err = a.mergeS3Info(merged.GetS3Info(), req)
	case dataflow.ConnectorType_GCS:
		merged.GcsInfo, err = a.mergeGcsInfo(merged.GetGcsInfo(), req)
	case dataflow.ConnectorType_AZURE_BLOB_STORAGE:
		merged.AzureBlobStorageInfo, err = a.mergeAzureBlobInfo(merged.GetAzureBlobStorageInfo(), req)
	default:
		err = bizerr.New(bizerr.ErrCodeParams, "unsupported downstream type")
	}
	if err != nil {
		return nil, err
	}
	return merged, nil
}

func (a *connectorChangefeedAdapter) connectorEditReqFromCreateReq(connector *dataflow.Connector,
	req *changefeedopenbp.EditChangefeedDownstreamConfigRequest,
) (*dataflow.ConnectorEditReq, error) {
	merged, err := a.mergeChangefeedEdit(connector, req)
	if err != nil {
		return nil, err
	}
	return &dataflow.ConnectorEditReq{
		ClusterId:            merged.GetClusterId(),
		ProjectId:            merged.GetProjectId(),
		OrgId:                merged.GetOrgId(),
		ConnectorId:          connector.GetConnectorId(),
		ConnectorType:        connector.GetConnectorType(),
		KafkaInfo:            merged.GetKafkaInfo(),
		MysqlInfo:            merged.GetMysqlInfo(),
		ServerlessTidbInfo:   merged.GetServerlessTidbInfo(),
		S3Info:               merged.GetS3Info(),
		GcsInfo:              merged.GetGcsInfo(),
		PulsarInfo:           merged.GetPulsarInfo(),
		AzureBlobStorageInfo: merged.GetAzureBlobStorageInfo(),
		Name:                 merged.GetName(),
		DryRun:               req.DryRun,
	}, nil
}

func connectorTypeFromDownstreamType(in changefeedopenbp.DownstreamType, required bool) (dataflow.ConnectorType, error) {
	switch in {
	case changefeedopenbp.DownstreamType_KAFKA:
		return dataflow.ConnectorType_KAFKA, nil
	case changefeedopenbp.DownstreamType_MYSQL:
		return dataflow.ConnectorType_MYSQL, nil
	case changefeedopenbp.DownstreamType_S3:
		return dataflow.ConnectorType_S3, nil
	case changefeedopenbp.DownstreamType_GCS:
		return dataflow.ConnectorType_GCS, nil
	case changefeedopenbp.DownstreamType_AZURE_BLOB:
		return dataflow.ConnectorType_AZURE_BLOB_STORAGE, nil
	case changefeedopenbp.DownstreamType_DOWNSTREAM_TYPE_UNSPECIFIED:
		if required {
			return 0, bizerr.New(bizerr.ErrCodeParams, "downstream_type is required")
		}
		return dataflow.ConnectorType_TOTAL, nil
	default:
		return 0, bizerr.New(bizerr.ErrCodeParams, "unsupported downstream_type")
	}
}

func downstreamTypeFromConnectorType(in dataflow.ConnectorType) changefeedopenbp.DownstreamType {
	switch in {
	case dataflow.ConnectorType_KAFKA:
		return changefeedopenbp.DownstreamType_KAFKA
	case dataflow.ConnectorType_MYSQL, dataflow.ConnectorType_TIDB, dataflow.ConnectorType_SERVERLESS_TIDB:
		return changefeedopenbp.DownstreamType_MYSQL
	case dataflow.ConnectorType_S3:
		return changefeedopenbp.DownstreamType_S3
	case dataflow.ConnectorType_GCS:
		return changefeedopenbp.DownstreamType_GCS
	case dataflow.ConnectorType_AZURE_BLOB_STORAGE:
		return changefeedopenbp.DownstreamType_AZURE_BLOB
	default:
		return changefeedopenbp.DownstreamType_DOWNSTREAM_TYPE_UNSPECIFIED
	}
}

func normalizeListPageSize(pageSize int32) uint32 {
	if pageSize <= 0 {
		return defaultListChangefeedsPageSize
	}
	if pageSize > 100 {
		pageSize = 100
	}
	return uint32(pageSize)
}

func hasNextPage(page, pageSize, total uint32) bool {
	if pageSize == 0 {
		return false
	}
	return (uint64(page)*uint64(pageSize) < uint64(total))
}

func connectorIDWithScope(connectorID uint64, scope changefeedScope) *dataflow.ConnectorID {
	return &dataflow.ConnectorID{
		ConnectorId: connectorID,
		ClusterId:   scope.clusterID(),
		OrgId:       scope.tenantID(),
		ProjectId:   scope.projectID(),
	}
}

func stringPtr(in string) *string {
	if in == "" {
		return nil
	}
	return &in
}

func (a *connectorChangefeedAdapter) mergeKafkaInfo(current *dataflow.KafkaInfo,
	req *changefeedopenbp.EditChangefeedDownstreamConfigRequest,
) (*dataflow.KafkaInfo, error) {
	merged := &dataflow.KafkaInfo{}
	if current != nil {
		merged = proto.Clone(current).(*dataflow.KafkaInfo)
	}
	if req.GetTableConfig() != nil {
		merged.TableConfig = toDataflowTableConfig(req.GetTableConfig())
	}
	if cfg := req.GetKafka(); cfg != nil {
		if cfg.GetBroker() != nil {
			merged.Broker = toDataflowKafkaBroker(cfg.GetBroker())
		}
		if cfg.GetAuthentication() != nil {
			merged.Authentication = toDataflowKafkaAuthentication(cfg.GetAuthentication())
		}
		if cfg.GetDataFormat() != nil {
			merged.DataFormat = toDataflowKafkaDataFormat(cfg.GetDataFormat())
		}
		if cfg.GetTopicPartitionConfig() != nil {
			merged.TopicPartitionConfig = toDataflowKafkaTopicPartitionConfig(cfg.GetTopicPartitionConfig())
		}
		if cfg.GetColumnSelectors() != nil {
			merged.ColumnSelectors = toDataflowKafkaColumnSelectors(cfg.GetColumnSelectors())
		}
	}
	return merged, nil
}

func (a *connectorChangefeedAdapter) mergeMySQLInfo(current *dataflow.MySQLInfo,
	req *changefeedopenbp.EditChangefeedDownstreamConfigRequest,
) (*dataflow.MySQLInfo, error) {
	merged := &dataflow.MySQLInfo{}
	if current != nil {
		merged = proto.Clone(current).(*dataflow.MySQLInfo)
	}
	if req.GetTableConfig() != nil {
		merged.TableConfig = toDataflowTableConfig(req.GetTableConfig())
	}
	if cfg := req.GetMysql(); cfg != nil && cfg.GetConnection() != nil {
		merged.Connection = toDataflowConnection(cfg.GetConnection())
	}
	return merged, nil
}

func (a *connectorChangefeedAdapter) mergeS3Info(current *dataflow.S3Info,
	req *changefeedopenbp.EditChangefeedDownstreamConfigRequest,
) (*dataflow.S3Info, error) {
	merged := &dataflow.S3Info{}
	if current != nil {
		merged = proto.Clone(current).(*dataflow.S3Info)
	}
	if req.GetTableConfig() != nil {
		merged.TableConfig = toDataflowTableConfig(req.GetTableConfig())
	}
	if cfg := req.GetS3(); cfg != nil {
		if cfg.GetEndpoint() != nil {
			merged.Endpoint = toDataflowS3Endpoint(cfg.GetEndpoint())
		}
		if cfg.GetDataFormat() != nil {
			dataFormat := toDataflowS3DataFormat(cfg.GetDataFormat())
			merged.DataFormat = dataFormat
		}
		if cfg.GetStorageFlushConfig() != nil {
			merged.StorageFlushConfig = toDataflowStorageFlushConfig(cfg.GetStorageFlushConfig())
		}
	}
	return merged, nil
}

func (a *connectorChangefeedAdapter) mergeGcsInfo(current *dataflow.GcsInfo,
	req *changefeedopenbp.EditChangefeedDownstreamConfigRequest,
) (*dataflow.GcsInfo, error) {
	merged := &dataflow.GcsInfo{}
	if current != nil {
		merged = proto.Clone(current).(*dataflow.GcsInfo)
	}
	if req.GetTableConfig() != nil {
		merged.TableConfig = toDataflowTableConfig(req.GetTableConfig())
	}
	if cfg := req.GetGcs(); cfg != nil {
		if cfg.GetEndpoint() != nil {
			merged.Endpoint = toDataflowGcsEndpoint(cfg.GetEndpoint())
		}
		if cfg.GetDataFormat() != nil {
			dataFormat := toDataflowS3DataFormat(cfg.GetDataFormat())
			merged.DataFormat = dataFormat
		}
		if cfg.GetStorageFlushConfig() != nil {
			merged.StorageFlushConfig = toDataflowStorageFlushConfig(cfg.GetStorageFlushConfig())
		}
	}
	return merged, nil
}

func (a *connectorChangefeedAdapter) mergeAzureBlobInfo(current *dataflow.AzureBlobStorageInfo,
	req *changefeedopenbp.EditChangefeedDownstreamConfigRequest,
) (*dataflow.AzureBlobStorageInfo, error) {
	merged := &dataflow.AzureBlobStorageInfo{}
	if current != nil {
		merged = proto.Clone(current).(*dataflow.AzureBlobStorageInfo)
	}
	if req.GetTableConfig() != nil {
		merged.TableConfig = toDataflowTableConfig(req.GetTableConfig())
	}
	if cfg := req.GetAzureBlob(); cfg != nil {
		if cfg.GetEndpoint() != nil {
			merged.Endpoint = toDataflowAzureBlobEndpoint(cfg.GetEndpoint())
		}
		if cfg.GetDataFormat() != nil {
			dataFormat := toDataflowS3DataFormat(cfg.GetDataFormat())
			merged.DataFormat = dataFormat
		}
		if cfg.GetStorageFlushConfig() != nil {
			merged.StorageFlushConfig = toDataflowStorageFlushConfig(cfg.GetStorageFlushConfig())
		}
	}
	return merged, nil
}