package changefeed

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

	ticdc_v1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/dataflow-service-ng/app/repo"
	"github.com/tidbcloud/dataflow-service-ng/app/utils"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	changefeedopenbp "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/openapi/v1beta2"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"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/structpb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

// --- Adapter setup and shared types ---

const defaultListChangefeedsPageSize uint32 = 20

var _ ChangefeedAdapter = (*consoleChangefeedAdapter)(nil)

type consoleChangefeedAdapter struct {
	consoleChangefeedAPI ConsoleChangefeedServer
	consoleSpecAPI       ConsoleChangefeedSpecServer
	changefeedRepo       repo.BizChangefeedRepo
}

func newConsoleChangefeedAdapter(
	consoleChangefeedAPI ConsoleChangefeedServer,
	consoleSpecAPI ConsoleChangefeedSpecServer,
	changefeedRepo repo.BizChangefeedRepo,
) ChangefeedAdapter {
	return &consoleChangefeedAdapter{
		consoleChangefeedAPI: consoleChangefeedAPI,
		consoleSpecAPI:       consoleSpecAPI,
		changefeedRepo:       changefeedRepo,
	}
}

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
}

// TODO(ng-cdc): migrate the internal changefeed specification code from the
// "rcu" suffix to "ccu". Until then, the OpenAPI layer exposes capacities as
// "ccu" while the console/internal layer still stores them as "rcu", so the
// two suffixes are converted at this adapter boundary.

const (
	specificationRCUSuffix = "rcu"
	specificationCCUSuffix = "ccu"
)

// toOpenAPICCU converts an internal changefeed specification code (e.g. "4rcu")
// to its OpenAPI representation ("4ccu").
func toOpenAPICCU(internal string) string {
	if internal == "" {
		return ""
	}
	return strings.TrimSuffix(internal, specificationRCUSuffix) + specificationCCUSuffix
}

// toConsoleSpecification converts an OpenAPI changefeed capacity (e.g. "4ccu")
// to the internal changefeed specification code ("4rcu").
func toConsoleSpecification(openAPI string) string {
	if openAPI == "" {
		return ""
	}
	return strings.TrimSuffix(openAPI, specificationCCUSuffix) + specificationRCUSuffix
}

// --- Public API operations ---

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

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

	consoleChangefeeds, err := a.consoleChangefeedAPI.ListChangefeeds(ctx, &dataflow.ChangefeedListReq{
		ClusterId:      tidbID,
		Page:           page,
		PerPage:        pageSize,
		ChangefeedType: consoleType,
	})
	if err != nil {
		return nil, err
	}
	if consoleChangefeeds == nil {
		consoleChangefeeds = &dataflow.Changefeeds{}
	}

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

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

	return out, nil
}

func (a *consoleChangefeedAdapter) CreateChangefeed(ctx context.Context,
	req *changefeedopenbp.CreateChangefeedRequest,
	tidbID uint64,
) (*changefeedopenbp.Changefeed, error) {
	createReq, err := a.toChangefeedCreateReq(req, tidbID)
	if err != nil {
		return nil, err
	}

	createdID, err := a.consoleChangefeedAPI.CreateChangefeed(ctx, createReq)
	if err != nil {
		return nil, err
	}
	if req.GetDryRun() {
		return &changefeedopenbp.Changefeed{}, nil
	}

	consoleChangefeed, err := a.consoleChangefeedAPI.GetChangefeed(ctx, createdID)
	if err != nil {
		return nil, err
	}
	return convertConsoleToPublicChangefeed(consoleChangefeed), nil
}

func (a *consoleChangefeedAdapter) DeleteChangefeed(ctx context.Context,
	req *changefeedopenbp.DeleteChangefeedRequest,
) (*emptypb.Empty, error) {
	changefeedID, err := a.changefeedID(ctx, req.GetChangefeedId())
	if err != nil {
		return nil, err
	}
	_, err = a.consoleChangefeedAPI.DeleteChangefeed(ctx, changefeedID)
	if err != nil {
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (a *consoleChangefeedAdapter) EditChangefeedDownstreamConfig(ctx context.Context,
	req *changefeedopenbp.EditChangefeedDownstreamConfigRequest,
) (*changefeedopenbp.Changefeed, error) {
	changefeedID, err := a.changefeedID(ctx, req.GetChangefeedId())
	if err != nil {
		return nil, err
	}
	consoleChangefeed, err := a.consoleChangefeedAPI.GetChangefeed(ctx, changefeedID)
	if err != nil {
		return nil, err
	}
	if consoleChangefeed == nil {
		return nil, bizerr.New(bizerr.ErrCodeGRPC, "changefeed not found")
	}

	editReq, err := a.changefeedEditReqFromCreateReq(consoleChangefeed, req)
	if err != nil {
		return nil, err
	}
	if _, err := a.consoleChangefeedAPI.EditChangefeed(ctx, editReq); err != nil {
		return nil, err
	}
	if req.GetDryRun() {
		return &changefeedopenbp.Changefeed{}, nil
	}

	updated, err := a.consoleChangefeedAPI.GetChangefeed(ctx, changefeedID)
	if err != nil {
		return nil, err
	}
	return convertConsoleToPublicChangefeed(updated), nil
}

func (a *consoleChangefeedAdapter) GetChangefeed(ctx context.Context,
	req *changefeedopenbp.GetChangefeedRequest,
) (*changefeedopenbp.Changefeed, error) {
	changefeedID, err := parseUintID(req.GetChangefeedId(), "changefeed_id")
	if err != nil {
		return nil, err
	}
	consoleChangefeed, err := a.consoleChangefeedAPI.GetChangefeed(ctx, &dataflow.ChangefeedID{
		ChangefeedId: changefeedID,
	})
	if err != nil {
		return nil, err
	}
	if consoleChangefeed == nil {
		return nil, bizerr.New(bizerr.ErrCodeGRPC, "changefeed not found")
	}
	if req.GetInvalidateCache() {
		if err := a.refreshRealtimeCheckpoint(ctx, consoleChangefeed); err != nil {
			return nil, err
		}
	}
	return convertConsoleToPublicChangefeed(consoleChangefeed), nil
}

func (a *consoleChangefeedAdapter) PauseChangefeed(ctx context.Context,
	req *changefeedopenbp.PauseChangefeedRequest,
) (*emptypb.Empty, error) {
	changefeedID, err := a.changefeedID(ctx, req.GetChangefeedId())
	if err != nil {
		return nil, err
	}
	_, err = a.consoleChangefeedAPI.PauseChangefeed(ctx, changefeedID)
	if err != nil {
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (a *consoleChangefeedAdapter) ResumeChangefeed(ctx context.Context,
	req *changefeedopenbp.ResumeChangefeedRequest,
) (*emptypb.Empty, error) {
	changefeedID, err := a.changefeedID(ctx, req.GetChangefeedId())
	if err != nil {
		return nil, err
	}
	_, err = a.consoleChangefeedAPI.ResumeChangefeed(ctx, changefeedID)
	if err != nil {
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (a *consoleChangefeedAdapter) ListCCUs(ctx context.Context,
	req *changefeedopenbp.ListCCUsRequest,
) (*changefeedopenbp.ListCCUsResponse, error) {
	_ = req
	resp, err := a.consoleSpecAPI.List(ctx, &dataflow.ChangefeedSpecListReq{})
	if err != nil {
		return nil, err
	}
	ccus := make([]*changefeedopenbp.ChangefeedCCU, 0, len(resp.GetItems()))
	for _, item := range resp.GetItems() {
		ccus = append(ccus, &changefeedopenbp.ChangefeedCCU{
			Name:        toOpenAPICCU(item.Name),
			DisplayName: fmt.Sprintf("%d CCUs", item.RcuNum),
			CcuNum:      item.RcuNum,
			RpsLimit:    item.RpsLimit,
		})
	}
	return &changefeedopenbp.ListCCUsResponse{Ccus: ccus}, nil
}

func (a *consoleChangefeedAdapter) ScaleChangefeed(ctx context.Context,
	req *changefeedopenbp.ScaleChangefeedRequest,
) (*changefeedopenbp.Changefeed, error) {
	changefeedID, err := a.changefeedID(ctx, req.GetChangefeedId())
	if err != nil {
		return nil, err
	}

	_, err = a.consoleChangefeedAPI.ScaleChangefeed(ctx, &dataflow.ChangefeedScaleReq{
		ChangefeedId:   changefeedID.GetChangefeedId(),
		ClusterId:      changefeedID.GetClusterId(),
		ChangefeedSpec: toConsoleSpecification(req.GetChangefeedCapacity()),
	})
	if err != nil {
		return nil, err
	}

	updated, err := a.consoleChangefeedAPI.GetChangefeed(ctx, changefeedID)
	if err != nil {
		return nil, err
	}
	return convertConsoleToPublicChangefeed(updated), nil
}

// --- Console response enrichment ---

func (a *consoleChangefeedAdapter) refreshRealtimeCheckpoint(
	ctx context.Context,
	consoleChangefeed *dataflow.Changefeed,
) error {
	logger := log.GetLogger(ctx).With(zap.Uint64("changefeed_id", consoleChangefeed.GetChangefeedId()))

	detail, err := a.consoleChangefeedAPI.GetChangefeedDetail(ctx, &dataflow.ChangefeedID{
		ChangefeedId: consoleChangefeed.GetChangefeedId(),
	})
	if err != nil {
		if bizErr, ok := err.(bizerr.BizError); ok && bizErr.RawErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil
		}
		return err
	}

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

	// 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
	}
	consoleChangefeed.CheckpointUtc = timestamppb.New(utils.GetTimeFromTS(kernelChangefeed.CheckpointTs))
	consoleChangefeed.CheckpointTso = kernelChangefeed.CheckpointTs
	return nil
}

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

	changefeedBytes, mErr := detail.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
}

// --- Console request construction and validation ---

func (a *consoleChangefeedAdapter) toChangefeedCreateReq(oreq *changefeedopenbp.CreateChangefeedRequest,
	tidbID uint64,
) (*dataflow.ChangefeedCreateReq, error) {
	cf := oreq.GetChangefeed()
	if cf == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "changefeed is required")
	}

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

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

	req := &dataflow.ChangefeedCreateReq{
		ClusterId:      tidbID,
		ChangefeedType: consoleType,
		Name:           cf.GetName(),
		ChangefeedSpec: toConsoleSpecification(cf.GetChangefeedCapacity()),
		NetworkInfo:    networkInfo,
		Internal:       false,
		DryRun:         oreq.GetDryRun(),
	}

	switch consoleType {
	case dataflow.ChangefeedType_KAFKA:
		req.KafkaInfo = toDataflowKafkaInfo(cf)
		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.ChangefeedType_MYSQL:
		req.MysqlInfo = toDataflowMySQLInfo(cf)
		if req.MysqlInfo == nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "mysql configuration is required")
		}
	case dataflow.ChangefeedType_S3:
		req.S3Info = toDataflowS3Info(cf)
		if req.S3Info == nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "s3 configuration is required")
		}
	case dataflow.ChangefeedType_GCS:
		req.GcsInfo = toDataflowGcsInfo(cf)
		if req.GcsInfo == nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "gcs configuration is required")
		}
	case dataflow.ChangefeedType_PULSAR:
		req.PulsarInfo = toDataflowPulsarInfo(cf)
		if req.PulsarInfo == nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "pulsar 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:
		return false
	default:
		return true
	}
}

func validateKafkaCreateInput(req *dataflow.ChangefeedCreateReq) 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 *consoleChangefeedAdapter) mergeChangefeedEdit(consoleChangefeed *dataflow.Changefeed,
	req *changefeedopenbp.EditChangefeedDownstreamConfigRequest,
) (*dataflow.ChangefeedCreateReq, error) {
	if consoleChangefeed == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "changefeed is required")
	}
	currentReq := consoleChangefeed.GetRequest()
	if currentReq == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "changefeed request is missing")
	}

	requestedType := req.GetDownstreamType()
	currentType := downstreamTypeFromConsoleType(consoleChangefeed.GetChangefeedType())
	if requestedType != currentType {
		return nil, bizerr.New(bizerr.ErrCodeParams, "downstream_type cannot be changed")
	}
	if err := validateEditDownstreamConfigType(req); err != nil {
		return nil, err
	}

	merged := proto.Clone(currentReq).(*dataflow.ChangefeedCreateReq)
	var err error
	switch consoleChangefeed.GetChangefeedType() {
	case dataflow.ChangefeedType_KAFKA:
		merged.KafkaInfo, err = a.mergeKafkaInfo(merged.GetKafkaInfo(), req)
	case dataflow.ChangefeedType_MYSQL, dataflow.ChangefeedType_TIDB:
		merged.MysqlInfo, err = a.mergeMySQLInfo(merged.GetMysqlInfo(), req)
	case dataflow.ChangefeedType_S3:
		merged.S3Info, err = a.mergeS3Info(merged.GetS3Info(), req)
	case dataflow.ChangefeedType_GCS:
		merged.GcsInfo, err = a.mergeGcsInfo(merged.GetGcsInfo(), req)
	case dataflow.ChangefeedType_PULSAR:
		merged.PulsarInfo, err = a.mergePulsarInfo(merged.GetPulsarInfo(), req)
	default:
		err = bizerr.New(bizerr.ErrCodeParams, "unsupported downstream type")
	}
	if err != nil {
		return nil, err
	}
	return merged, nil
}

func (a *consoleChangefeedAdapter) changefeedEditReqFromCreateReq(consoleChangefeed *dataflow.Changefeed,
	req *changefeedopenbp.EditChangefeedDownstreamConfigRequest,
) (*dataflow.ChangefeedEditReq, error) {
	merged, err := a.mergeChangefeedEdit(consoleChangefeed, req)
	if err != nil {
		return nil, err
	}
	return &dataflow.ChangefeedEditReq{
		ClusterId:      merged.GetClusterId(),
		ChangefeedId:   consoleChangefeed.GetChangefeedId(),
		ChangefeedType: consoleChangefeed.GetChangefeedType(),
		KafkaInfo:      merged.GetKafkaInfo(),
		MysqlInfo:      merged.GetMysqlInfo(),
		S3Info:         merged.GetS3Info(),
		GcsInfo:        merged.GetGcsInfo(),
		PulsarInfo:     merged.GetPulsarInfo(),
		Name:           merged.GetName(),
		DryRun:         req.GetDryRun(),
	}, nil
}

// --- Type, pagination, and request helpers ---

func consoleTypeFromDownstreamType(in changefeedopenbp.DownstreamType, required bool) (dataflow.ChangefeedType, error) {
	switch in {
	case changefeedopenbp.DownstreamType_KAFKA:
		return dataflow.ChangefeedType_KAFKA, nil
	case changefeedopenbp.DownstreamType_MYSQL:
		return dataflow.ChangefeedType_MYSQL, nil
	case changefeedopenbp.DownstreamType_S3:
		return dataflow.ChangefeedType_S3, nil
	case changefeedopenbp.DownstreamType_GCS:
		return dataflow.ChangefeedType_GCS, nil
	case changefeedopenbp.DownstreamType_PULSAR:
		return dataflow.ChangefeedType_PULSAR, nil
	case changefeedopenbp.DownstreamType_DOWNSTREAM_TYPE_UNSPECIFIED:
		if required {
			return 0, bizerr.New(bizerr.ErrCodeParams, "downstream_type is required")
		}
		return dataflow.ChangefeedType_TOTAL, nil
	default:
		return 0, bizerr.New(bizerr.ErrCodeParams, "unsupported downstream_type")
	}
}

func downstreamTypeFromConsoleType(in dataflow.ChangefeedType) changefeedopenbp.DownstreamType {
	switch in {
	case dataflow.ChangefeedType_KAFKA:
		return changefeedopenbp.DownstreamType_KAFKA
	case dataflow.ChangefeedType_MYSQL, dataflow.ChangefeedType_TIDB:
		return changefeedopenbp.DownstreamType_MYSQL
	case dataflow.ChangefeedType_S3:
		return changefeedopenbp.DownstreamType_S3
	case dataflow.ChangefeedType_GCS:
		return changefeedopenbp.DownstreamType_GCS
	case dataflow.ChangefeedType_PULSAR:
		return changefeedopenbp.DownstreamType_PULSAR
	default:
		return changefeedopenbp.DownstreamType_DOWNSTREAM_TYPE_UNSPECIFIED
	}
}

func validateEditDownstreamConfigType(req *changefeedopenbp.EditChangefeedDownstreamConfigRequest) error {
	var configType changefeedopenbp.DownstreamType
	switch req.GetDownstreamConfig().(type) {
	case nil:
		return nil
	case *changefeedopenbp.EditChangefeedDownstreamConfigRequest_Kafka:
		configType = changefeedopenbp.DownstreamType_KAFKA
	case *changefeedopenbp.EditChangefeedDownstreamConfigRequest_Mysql:
		configType = changefeedopenbp.DownstreamType_MYSQL
	case *changefeedopenbp.EditChangefeedDownstreamConfigRequest_S3:
		configType = changefeedopenbp.DownstreamType_S3
	case *changefeedopenbp.EditChangefeedDownstreamConfigRequest_Gcs:
		configType = changefeedopenbp.DownstreamType_GCS
	case *changefeedopenbp.EditChangefeedDownstreamConfigRequest_Pulsar:
		configType = changefeedopenbp.DownstreamType_PULSAR
	default:
		return bizerr.New(bizerr.ErrCodeParams, "unsupported downstream configuration")
	}
	if configType != req.GetDownstreamType() {
		return bizerr.New(bizerr.ErrCodeParams, "downstream configuration does not match downstream_type")
	}
	return nil
}

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 (a *consoleChangefeedAdapter) changefeedID(
	ctx context.Context,
	rawChangefeedID string,
) (*dataflow.ChangefeedID, error) {
	changefeedID, err := parseUintID(rawChangefeedID, "changefeed_id")
	if err != nil {
		return nil, err
	}
	changefeed, repoErr := a.changefeedRepo.Load(ctx, changefeedID)
	if repoErr != nil {
		return nil, repoErr
	}
	if changefeed == nil {
		return nil, bizerr.New(bizerr.ErrCodeGRPC, "changefeed not found")
	}
	return &dataflow.ChangefeedID{
		ChangefeedId: changefeedID,
		ClusterId:    changefeed.ClusterID.Uint64(),
	}, nil
}

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

// --- Downstream-specific edit merge helpers ---

func (a *consoleChangefeedAdapter) 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 *consoleChangefeedAdapter) 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 *consoleChangefeedAdapter) 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 *consoleChangefeedAdapter) 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 *consoleChangefeedAdapter) mergePulsarInfo(current *dataflow.PulsarInfo,
	req *changefeedopenbp.EditChangefeedDownstreamConfigRequest,
) (*dataflow.PulsarInfo, error) {
	merged := &dataflow.PulsarInfo{}
	if current != nil {
		merged = proto.Clone(current).(*dataflow.PulsarInfo)
	}
	if req.GetTableConfig() != nil {
		merged.TableConfig = toDataflowTableConfig(req.GetTableConfig())
	}
	if cfg := req.GetPulsar(); cfg != nil {
		updated := toDataflowPulsarConfig(cfg)
		if updated.GetAuthentication() != nil && merged.GetAuthentication() != nil {
			if updated.Authentication.Password == "" {
				updated.Authentication.Password = merged.Authentication.Password
			}
			if updated.Authentication.Token == "" {
				updated.Authentication.Token = merged.Authentication.Token
			}
		}
		updated.TableConfig = merged.TableConfig
		updated.StartPosition = merged.StartPosition
		merged = updated
	}
	return merged, nil
}