package impl

import (
	"context"
	"encoding/json"
	"fmt"
	"reflect"
	"strings"
	"time"

	"github.com/elliotchance/pie/v2"
	"github.com/imdario/mergo"
	ticdc_v1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	appconfig "github.com/tidbcloud/dataflow-service-ng/app/config"
	"github.com/tidbcloud/dataflow-service-ng/app/models/config_manager"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	"github.com/tidbcloud/dataflow-service-ng/app/models/ticdc"
	"github.com/tidbcloud/dataflow-service-ng/app/repo"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/app/utils"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	regionalapi "github.com/tidbcloud/dataflow-service-ng/pkg/regional-api"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/structpb"
)

const cdcErrorCodeAlreadyExists = "ErrChangeFeedAlreadyExists"

type changefeedModelRegionalImpl struct {
	changefeedCli     regionalapi.ChangefeedClient
	cdcClusterCli     regionalapi.TiCDCClusterClient
	transProvider     db.TransProvider
	changefeedRepo    repo.TiCDCChangefeedRepo
	ticdcClusterRepo  repo.TiCDCClusterRepo
	bizChangefeedRepo repo.BizChangefeedRepo
	configManager     config_manager.ChangefeedConfigManager
}

func NewChangefeedRegionalModel(
	transProvider db.TransProvider,
	changefeedCli regionalapi.ChangefeedClient,
	cdcClusterCli regionalapi.TiCDCClusterClient,
	changefeedRepo repo.TiCDCChangefeedRepo,
	ticdcClusterRepo repo.TiCDCClusterRepo,
	bizChangefeedRepo repo.BizChangefeedRepo,
	configManager config_manager.ChangefeedConfigManager,
) ticdc.ChangefeedModel {
	return &changefeedModelRegionalImpl{
		changefeedCli:     changefeedCli,
		cdcClusterCli:     cdcClusterCli,
		transProvider:     transProvider,
		changefeedRepo:    changefeedRepo,
		ticdcClusterRepo:  ticdcClusterRepo,
		bizChangefeedRepo: bizChangefeedRepo,
		configManager:     configManager,
	}
}

func newRegionFromTiCDCCluster(ticdcClusterEnt *entity.TiCDCCluster) *regionalapi.Region {
	return &regionalapi.Region{
		SharedPoolID: ticdcClusterEnt.SharedPoolID,
	}
}

// CreateChangefeed runs a 3-phase, DAG-retry-driven creation flow:
//
//	Phase 1 (short tx): persistCreateIntent — load existing pending row by
//	  deterministic kernel ID, or insert a new row with KernelStatus=pending
//	  plus its config-center records. The row is committed BEFORE we talk to
//	  regional so that pod crashes / kernel timeouts do not leave kernel-side
//	  orphans without a DB pointer.
//
//	Phase 2 (no tx): createOrVerifyKernelChangefeed — call regional Create.
//	  On error we fall back to the existing-changefeed short-circuit and the
//	  list-based idempotency probe (RCA_Changefeed_Creation_Mismatch.md).
//
//	Phase 3 (single Save): flip KernelStatus from pending to active once we
//	  know the kernel owns the changefeed.
//
// If Phase 2 fails, the row stays pending and the DAG action retries. The
// deterministic kernel ID + the load-or-create logic in Phase 1 makes every
// subsequent attempt idempotent.
func (m *changefeedModelRegionalImpl) CreateChangefeed(
	ctx context.Context, req *domain.CreateChangefeedReq,
) (*domain.CreateChangefeedResp, bizerr.BizError) {
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, req.TiCDCClusterID)
	if err != nil {
		return nil, err.Chain("query ticdc cluster entity")
	}

	changefeedEnt, err := m.persistCreateIntent(ctx, req, ticdcClusterEnt)
	if err != nil {
		return nil, err.Chain("persist create intent")
	}
	resp := &domain.CreateChangefeedResp{
		TiCDCClusterID: req.TiCDCClusterID,
		ChangefeedID:   changefeedEnt.ID.Uint64(),
	}
	if changefeedEnt.KernelStatus == entity.TiCDCChangefeedKernelStatusActive {
		// Already converged; do not re-issue regional Create.
		return resp, nil
	}

	if err := m.createOrVerifyKernelChangefeed(ctx, req.Config, ticdcClusterEnt, changefeedEnt); err != nil {
		// Row stays pending; DAG action retries.
		return nil, err
	}

	changefeedEnt.KernelStatus = entity.TiCDCChangefeedKernelStatusActive
	if err := m.changefeedRepo.Save(ctx, changefeedEnt); err != nil {
		return nil, err.Chain("activate changefeed entity")
	}
	return resp, nil
}

// persistCreateIntent returns the durable ticdc_changefeed row for this
// create request, creating it only on first call. Subsequent calls (DAG
// retries, pod restarts) load the existing row by KernelChangefeedID — which
// is deterministic per biz_changefeed and uniquely indexed — so we never
// double-insert.
func (m *changefeedModelRegionalImpl) persistCreateIntent(
	ctx context.Context, req *domain.CreateChangefeedReq, ticdcClusterEnt *entity.TiCDCCluster,
) (*entity.TiCDCChangefeed, bizerr.BizError) {
	existing, err := m.changefeedRepo.Get(ctx,
		func(qs entity.TiCDCChangefeedQuerySet) entity.TiCDCChangefeedQuerySet {
			return qs.KernelChangefeedIDEq(req.KernelChangefeedID).DeletedAtIsNull()
		})
	if err == nil {
		return existing, nil
	}
	if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return nil, err.Chain("look up existing changefeed entity")
	}

	changefeedEnt, err := m.assembleChangefeedEntity(ctx, req, ticdcClusterEnt)
	if err != nil {
		return nil, err.Chain("assemble changefeed entity")
	}
	changefeedEnt.KernelStatus = entity.TiCDCChangefeedKernelStatusPending
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		if err := m.changefeedRepo.Save(ctx, changefeedEnt); err != nil {
			return err.Chain("save changefeed entity")
		}
		if _, err := m.configManager.Create(ctx, &domain.CreateChangefeedConfigReq{
			ChangefeedID:   changefeedEnt.ID.Uint64(),
			DownstreamType: changefeedEnt.DownStreamType,
		}); err != nil {
			return err.Chain("create changefeed config")
		}
		return nil
	})
	if transErr != nil {
		return nil, convertTransactionError(transErr)
	}
	return changefeedEnt, nil
}

// createOrVerifyKernelChangefeed is the non-transactional kernel-call phase.
// Treats "already exists" plus a successful list-based idempotency probe as
// success so timeout / context-canceled responses don't fail the create when
// the kernel actually completed.
func (m *changefeedModelRegionalImpl) createOrVerifyKernelChangefeed(
	ctx context.Context,
	userDefinedConfig *domain.TiCDCChangefeedConfig,
	ticdcClusterEnt *entity.TiCDCCluster,
	changefeedEnt *entity.TiCDCChangefeed,
) bizerr.BizError {
	uriConfig, replicaConfig, err := m.assembleKernelChangefeedConfig(ctx, changefeedEnt, userDefinedConfig)
	if err != nil {
		return err
	}
	var startTS uint64
	if userDefinedConfig.Config.StartTs != nil {
		startTS = *userDefinedConfig.Config.StartTs
	}

	if err := m.createChangefeedByRegional(
		ctx, startTS, uriConfig, replicaConfig, ticdcClusterEnt, changefeedEnt); err != nil {
		if strings.Contains(err.Error(), cdcErrorCodeAlreadyExists) {
			log.GetLogger(ctx).Warn("create changefeed already exists, skip the error",
				zap.String("kernel_changefeed_id", changefeedEnt.KernelChangefeedID),
				zap.Uint64("changefeed_id", changefeedEnt.ID.Uint64()), zap.Error(err))
			return nil
		}
		// Idempotency probe for the timeout / context-canceled case: regional
		// may have actually created the changefeed in the kernel but our
		// response was lost. See RCA_Changefeed_Creation_Mismatch.md.
		if exists, probeErr := m.kernelChangefeedExists(ctx, changefeedEnt, ticdcClusterEnt); probeErr != nil {
			log.GetLogger(ctx).Warn("kernel existence probe failed after regional create error, propagating original error",
				zap.String("kernel_changefeed_id", changefeedEnt.KernelChangefeedID),
				zap.Error(probeErr))
		} else if exists {
			log.GetLogger(ctx).Warn("regional create returned error but kernel already has the changefeed; treating as success",
				zap.String("kernel_changefeed_id", changefeedEnt.KernelChangefeedID),
				zap.Error(err))
			return nil
		}
		log.GetLogger(ctx).Warn("create changefeed by regional error",
			zap.String("kernel_changefeed_id", changefeedEnt.KernelChangefeedID),
			zap.Uint64("changefeed_id", changefeedEnt.ID.Uint64()),
			zap.Error(err))
		return bizerr.New(appconfig.ErrCodeCreateKernelChangefeedFailed, "%s", err.ErrMsg())
	}
	return nil
}

// kernelChangefeedExists probes the kernel via ListChangefeeds to
// definitively answer "is this changefeed currently in the kernel?". List
// is used (not Get) because the regional service forwards kernel errors as
// opaque non-2xx bodies, which makes structured "not found" detection on Get
// unreliable. With list, presence/absence is unambiguous, and any error
// returned reflects an actual transport/regional failure rather than a
// missing-record signal.
func (m *changefeedModelRegionalImpl) kernelChangefeedExists(
	ctx context.Context, changefeedEnt *entity.TiCDCChangefeed, ticdcClusterEnt *entity.TiCDCCluster,
) (bool, bizerr.BizError) {
	region := newRegionFromTiCDCCluster(ticdcClusterEnt)
	resp, err := m.changefeedCli.ListChangefeeds(ctx, region, &ticdc_v1.ListChangefeedsRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
		TidbClusterId:  fmt.Sprintf("%d", changefeedEnt.UpStreamClusterID.Uint64()),
	})
	if err != nil {
		return false, err
	}
	for _, cf := range resp.Items {
		if cf.Id == changefeedEnt.KernelChangefeedID {
			return true, nil
		}
	}
	return false, nil
}

func (m *changefeedModelRegionalImpl) assembleKernelChangefeedConfig(
	ctx context.Context, changefeedEnt *entity.TiCDCChangefeed, userDefinedConfig *domain.TiCDCChangefeedConfig,
) (domain.TiCDCChangefeedURIConfig, *ticdc_v1.ReplicaConfig, bizerr.BizError) {
	configCenterDefinedConfig, err := m.configManager.Get(ctx, &domain.GetChangefeedConfigReq{
		ChangefeedID: changefeedEnt.ID.Uint64(),
	})
	if err != nil {
		return nil, nil, err.Chain("get changefeed config")
	}
	userDefinedReplicaConfig := m.normalizeChangefeedReplicaConfig(userDefinedConfig.Config.ReplicaConfig)
	uriConfig, err := m.overrideChangefeedURIConfig(
		userDefinedConfig.URIConfig, configCenterDefinedConfig.URIConfiguration)
	if err != nil {
		return nil, nil, err.Chain("override changefeed uri config")
	}
	replicaConfig, err := m.overrideChangefeedReplicaConfig(
		userDefinedReplicaConfig, configCenterDefinedConfig.Configuration)
	if err != nil {
		return nil, nil, err.Chain("override changefeed replica config")
	}
	return uriConfig, replicaConfig, nil
}

func (m *changefeedModelRegionalImpl) createChangefeedByRegional(
	ctx context.Context,
	startTS uint64,
	uriConfig domain.TiCDCChangefeedURIConfig,
	replicaConfig *ticdc_v1.ReplicaConfig,
	ticdcClusterEnt *entity.TiCDCCluster,
	changefeedEnt *entity.TiCDCChangefeed,
) bizerr.BizError {
	region := newRegionFromTiCDCCluster(ticdcClusterEnt)
	regionalReq := &ticdc_v1.CreateChangefeedRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
		TidbClusterId:  fmt.Sprintf("%d", changefeedEnt.UpStreamClusterID.Uint64()),
		ChangefeedId:   changefeedEnt.KernelChangefeedID,
		Changefeed: &ticdc_v1.ChangefeedConfig{
			ChangefeedId:  changefeedEnt.KernelChangefeedID,
			StartTs:       startTS,
			SinkUri:       uriConfig.GenerateURI(),
			ReplicaConfig: replicaConfig,
		},
	}
	_, err := m.changefeedCli.CreateChangefeed(ctx, region, regionalReq)
	return err
}

func (m *changefeedModelRegionalImpl) assembleChangefeedEntity(
	ctx context.Context, req *domain.CreateChangefeedReq, ticdcClusterEnt *entity.TiCDCCluster,
) (*entity.TiCDCChangefeed, bizerr.BizError) {
	bizChangefeedEnt, err := m.bizChangefeedRepo.Load(ctx, req.BizChangefeedID)
	if err != nil {
		return nil, err.Chain("load biz changefeed entity")
	}
	changefeedEnt := &entity.TiCDCChangefeed{
		KernelChangefeedID: req.KernelChangefeedID,
		TenantID:           bizChangefeedEnt.TenantID,
		TiCDCClusterID:     ticdcClusterEnt.ID,
		UpStreamClusterID:  entity.ID(req.TiDBClusterID),
		BizChangefeedID:    entity.ID(req.BizChangefeedID),
		DownStreamType:     req.DownStreamType,
	}
	return changefeedEnt, nil
}

func (m *changefeedModelRegionalImpl) overrideChangefeedReplicaConfig(
	userDefinedReplicaConfig *ticdc_v1.ReplicaConfig,
	configCenterDefinedReplicaConfig string,
) (*ticdc_v1.ReplicaConfig, bizerr.BizError) {
	dstReplicaConfig := new(ticdc_v1.ReplicaConfig)
	if err := protojson.Unmarshal([]byte(configCenterDefinedReplicaConfig), dstReplicaConfig); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "unmarshal replica config")
	}
	if err := mergo.Merge(dstReplicaConfig, userDefinedReplicaConfig,
		mergo.WithOverride, mergo.WithTransformers(&utils.BoolMergoTransFormer{})); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "override replica config")
	}
	return dstReplicaConfig, nil
}

func (m *changefeedModelRegionalImpl) overrideChangefeedURIConfig(
	userDefinedURIConfig domain.TiCDCChangefeedURIConfig,
	configCenterDefinedURIConfig string,
) (domain.TiCDCChangefeedURIConfig, bizerr.BizError) {
	dstUriConfig := reflect.New(reflect.TypeOf(userDefinedURIConfig).Elem()).
		Interface().(domain.TiCDCChangefeedURIConfig)
	uriConfigDecoder := json.NewDecoder(strings.NewReader(configCenterDefinedURIConfig))
	uriConfigDecoder.DisallowUnknownFields()
	if err := uriConfigDecoder.Decode(dstUriConfig); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "unmarshal uri config")
	}
	if err := mergo.Merge(dstUriConfig, userDefinedURIConfig,
		mergo.WithOverride, mergo.WithTransformers(&utils.BoolMergoTransFormer{})); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "override uri config")
	}
	return dstUriConfig, nil
}

func (m *changefeedModelRegionalImpl) normalizeChangefeedReplicaConfig(
	conf *clusterproxy.ReplicaConfig,
) *ticdc_v1.ReplicaConfig {
	regionalConfig := m.normalizeBasicChangefeedReplicaConfig(conf)
	if conf.Filter != nil {
		regionalConfig.Filter = m.convertRegionalChangefeedFilter(conf.Filter)
	}
	if conf.Mounter != nil {
		regionalConfig.Mounter = &ticdc_v1.ReplicaConfig_MounterConfig{
			WorkerNum: int32(conf.Mounter.WorkerNum),
		}
	}
	if conf.Sink != nil {
		regionalConfig.Sink = m.convertRegionalChangefeedSink(conf.Sink)
	}
	if conf.Consistent != nil {
		regionalConfig.Consistent = &ticdc_v1.ReplicaConfig_ConsistentConfig{
			Level:         conf.Consistent.Level,
			MaxLogSize:    uint64(conf.Consistent.MaxLogSize),
			FlushInterval: uint64(conf.Consistent.FlushIntervalInMs),
			Storage:       conf.Consistent.Storage,
		}
	}
	return regionalConfig
}

func (m *changefeedModelRegionalImpl) normalizeBasicChangefeedReplicaConfig(
	conf *clusterproxy.ReplicaConfig,
) *ticdc_v1.ReplicaConfig {
	regionalConfig := &ticdc_v1.ReplicaConfig{
		MemoryQuota:           conf.MemoryQuota,
		SqlMode:               conf.SQLMode,
		CaseSensitive:         conf.CaseSensitive,
		EnableOldValue:        conf.EnableOldValue,
		ForceReplicate:        conf.ForceReplicate,
		IgnoreIneligibleTable: conf.IgnoreIneligibleTable,
		CheckGcSafePoint:      conf.CheckGCSafePoint,
		EnableSyncPoint:       conf.EnableSyncPoint,
		BdrMode:               conf.BDRMode,
	}
	return regionalConfig
}

func (m *changefeedModelRegionalImpl) convertRegionalChangefeedFilter(
	filterConf *clusterproxy.FilterConfig,
) *ticdc_v1.ReplicaConfig_FilterConfig {
	regionalConfig := &ticdc_v1.ReplicaConfig_FilterConfig{
		Rules:            filterConf.Rules,
		IgnoreTxnStartTs: filterConf.IgnoreTxnStartTs,
		EventFilters:     make([]*ticdc_v1.ReplicaConfig_EventFilterRule, 0, len(filterConf.EventFilters)),
	}
	for _, eventFilter := range filterConf.EventFilters {
		regionalEventFilter := &ticdc_v1.ReplicaConfig_EventFilterRule{
			Matcher:                  eventFilter.Matcher,
			IgnoreEvent:              eventFilter.IgnoreEvent,
			IgnoreSql:                eventFilter.IgnoreSQL,
			IgnoreInsertValueExpr:    eventFilter.IgnoreInsertValueExpr,
			IgnoreUpdateNewValueExpr: eventFilter.IgnoreUpdateNewValueExpr,
			IgnoreUpdateOldValueExpr: eventFilter.IgnoreUpdateOldValueExpr,
			IgnoreDeleteValueExpr:    eventFilter.IgnoreDeleteValueExpr,
		}
		regionalConfig.EventFilters = append(regionalConfig.EventFilters, regionalEventFilter)
	}
	return regionalConfig
}

func (m *changefeedModelRegionalImpl) convertRegionalChangefeedSink(
	sinkConf *clusterproxy.SinkConfigV2,
) *ticdc_v1.ReplicaConfig_SinkConfig {
	regionalConfig := &ticdc_v1.ReplicaConfig_SinkConfig{
		Protocol:              sinkConf.Protocol,
		SchemaRegistry:        sinkConf.SchemaRegistry,
		TransactionAtomicity:  sinkConf.TxnAtomicity,
		EncoderConcurrency:    uint32(sinkConf.EncoderConcurrency),
		Terminator:            sinkConf.Terminator,
		DateSeparator:         sinkConf.DateSeparator,
		Dispatchers:           make([]*ticdc_v1.ReplicaConfig_DispatchRule, 0, len(sinkConf.DispatchRules)),
		ColumnSelectors:       make([]*ticdc_v1.ReplicaConfig_ColumnSelector, 0, len(sinkConf.ColumnSelectors)),
		DebeziumDisableSchema: sinkConf.DebeziumDisableSchema,
	}
	if sinkConf.CSVConfig != nil {
		regionalConfig.Csv = &ticdc_v1.ReplicaConfig_CSVConfig{
			Delimiter:            sinkConf.CSVConfig.Delimiter,
			Quote:                sinkConf.CSVConfig.Quote,
			Null:                 sinkConf.CSVConfig.NullString,
			BinaryEncodingMethod: sinkConf.CSVConfig.BinaryEncodingMethod,
			IncludeCommitTs:      sinkConf.CSVConfig.IncludeCommitTs,
		}
	}
	for _, dispatcherRule := range sinkConf.DispatchRules {
		regionalConfig.Dispatchers = append(regionalConfig.Dispatchers, &ticdc_v1.ReplicaConfig_DispatchRule{
			Matcher:   dispatcherRule.Matcher,
			Topic:     dispatcherRule.Topic,
			Partition: dispatcherRule.Partition,
			Index:     dispatcherRule.IndexName,
			Columns:   dispatcherRule.Columns,
		})
	}
	for _, columnSelector := range sinkConf.ColumnSelectors {
		regionalConfig.ColumnSelectors = append(regionalConfig.ColumnSelectors,
			&ticdc_v1.ReplicaConfig_ColumnSelector{
				Matcher: columnSelector.Matcher,
				Columns: columnSelector.Columns,
			})
	}
	if sinkConf.DebeziumConfig != nil {
		regionalConfig.Debezium = &ticdc_v1.ReplicaConfig_DebeziumConfig{
			OutputOldValue: sinkConf.DebeziumConfig.OutputOldValue,
		}
	}
	if sinkConf.PulsarConfig != nil {
		regionalConfig.PulsarConfig = m.convertRegionalPulsarSink(sinkConf.PulsarConfig)
	}
	return regionalConfig
}

func (m *changefeedModelRegionalImpl) convertRegionalPulsarSink(
	pulsarConfig *clusterproxy.PulsarConfig,
) *ticdc_v1.ReplicaConfig_PulsarConfig {
	regionalConfig := &ticdc_v1.ReplicaConfig_PulsarConfig{
		AuthenticationToken:     pulsarConfig.AuthenticationToken,
		BasicUserName:           pulsarConfig.BasicUserName,
		BasicPassword:           pulsarConfig.BasicPassword,
		CompressionType:         pulsarConfig.CompressionType,
		ConnectionTimeout:       pulsarConfig.ConnectionTimeout,
		OperationTimeout:        pulsarConfig.OperationTimeout,
		SendTimeout:             pulsarConfig.SendTimeout,
		BatchingMaxMessages:     pulsarConfig.BatchingMaxMessages,
		BatchingMaxPublishDelay: pulsarConfig.BatchingMaxPublishDelay,
	}
	return regionalConfig
}

func (m *changefeedModelRegionalImpl) DeleteChangefeed(
	ctx context.Context, req *domain.DeleteChangefeedReq,
) (*domain.DeleteChangefeedResp, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	changefeedEnt, err := m.changefeedRepo.Load(ctx, req.ChangefeedID)
	if err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			logger.Info("changefeed is not found", zap.Uint64("changefeed_id", req.ChangefeedID))
			return &domain.DeleteChangefeedResp{}, nil
		}
		return nil, err.Chain("load changefeed entity")
	}
	if changefeedEnt.DeletedAt != nil {
		logger.Info("changefeed is already deleted",
			zap.Uint64("changefeed_id", req.ChangefeedID), zap.Time("deleted_at", *changefeedEnt.DeletedAt))
		return &domain.DeleteChangefeedResp{}, nil
	}
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, changefeedEnt.TiCDCClusterID.Uint64())
	if err != nil {
		return nil, err.Chain("load ticdc cluster entity")
	}

	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.doDeleteChangefeed(ctx, changefeedEnt, ticdcClusterEnt)
	})
	if transErr != nil {
		return nil, convertTransactionError(transErr)
	}
	return &domain.DeleteChangefeedResp{}, nil
}

func (m *changefeedModelRegionalImpl) doDeleteChangefeed(
	ctx context.Context, changefeedEnt *entity.TiCDCChangefeed, ticdcClusterEnt *entity.TiCDCCluster,
) bizerr.BizError {
	deletedAt := time.Now()
	changefeedEnt.DeletedAt = &deletedAt
	if err := m.changefeedRepo.Save(ctx, changefeedEnt); err != nil {
		return err.Chain("save changefeed entity")
	}
	if err := m.deleteChangefeedByRegional(ctx, changefeedEnt, ticdcClusterEnt); err != nil {
		return err.Chain("delete changefeed by regional")
	}
	return nil
}

func (m *changefeedModelRegionalImpl) deleteChangefeedByRegional(
	ctx context.Context, changefeedEnt *entity.TiCDCChangefeed, ticdcClusterEnt *entity.TiCDCCluster,
) bizerr.BizError {
	region := newRegionFromTiCDCCluster(ticdcClusterEnt)
	regionalReq := &ticdc_v1.DeleteChangefeedRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
		TidbClusterId:  fmt.Sprintf("%d", changefeedEnt.UpStreamClusterID.Uint64()),
		ChangefeedId:   changefeedEnt.KernelChangefeedID,
	}
	return m.changefeedCli.DeleteChangefeed(ctx, region, regionalReq)
}

func (m *changefeedModelRegionalImpl) ModifyChangefeedConfig(
	ctx context.Context, req *domain.ModifyChangefeedConfigReq,
) (*domain.ModifyChangefeedConfigResp, bizerr.BizError) {
	changefeedEnt, err := m.changefeedRepo.Load(ctx, req.ChangefeedID)
	if err != nil {
		return nil, err.Chain("load changefeed entity")
	}
	if changefeedEnt.KernelMetrics.FeedState != entity.ChangefeedKernelStatusStopped &&
		changefeedEnt.KernelMetrics.FeedState != entity.ChangefeedKernelStatusFailed {
		return nil, bizerr.New(bizerr.ErrCodeParams,
			"changefeed with status %s is not allowed to update", changefeedEnt.KernelMetrics.FeedState)
	}
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, changefeedEnt.TiCDCClusterID.Uint64())
	if err != nil {
		return nil, err.Chain("load ticdc cluster entity")
	}
	if err = m.doModifyChangefeedConfig(ctx, req.Config, ticdcClusterEnt, changefeedEnt); err != nil {
		return nil, err.Chain("modify changefeed config")
	}
	return &domain.ModifyChangefeedConfigResp{}, nil
}

func (m *changefeedModelRegionalImpl) doModifyChangefeedConfig(
	ctx context.Context,
	userDefinedConfig *domain.TiCDCChangefeedConfig,
	ticdcClusterEnt *entity.TiCDCCluster,
	changefeedEnt *entity.TiCDCChangefeed,
) bizerr.BizError {
	kernelChangefeed, err := m.getChangefeedFromRegional(ctx, changefeedEnt, ticdcClusterEnt)
	if err != nil {
		return err.Chain("get kernel changefeed")
	}
	uriConfig, patchReplicaConfig, err := m.assembleKernelChangefeedConfig(ctx, changefeedEnt, userDefinedConfig)
	if err != nil {
		return err
	}

	// patch user defined config to current changefeed config
	dstReplicaConfig := kernelChangefeed.Config
	if err := mergo.Merge(dstReplicaConfig, patchReplicaConfig,
		mergo.WithOverride, mergo.WithOverrideEmptySlice, mergo.WithTransformers(&utils.BoolMergoTransFormer{}),
	); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "patch replica config")
	}
	if err = m.modifyChangefeedConfigByRegional(
		ctx, uriConfig, dstReplicaConfig, ticdcClusterEnt, changefeedEnt); err != nil {
		return err.Chain("modify changefeed config by regional")
	}
	return nil
}

func (m *changefeedModelRegionalImpl) getChangefeedFromRegional(
	ctx context.Context, changefeedEnt *entity.TiCDCChangefeed, ticdcClusterEnt *entity.TiCDCCluster,
) (*ticdc_v1.Changefeed, bizerr.BizError) {
	region := newRegionFromTiCDCCluster(ticdcClusterEnt)
	regionalReq := &ticdc_v1.GetChangefeedRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
		TidbClusterId:  fmt.Sprintf("%d", changefeedEnt.UpStreamClusterID.Uint64()),
		ChangefeedId:   changefeedEnt.KernelChangefeedID,
	}
	return m.changefeedCli.GetChangefeed(ctx, region, regionalReq)
}

func (m *changefeedModelRegionalImpl) modifyChangefeedConfigByRegional(
	ctx context.Context,
	uriConfig domain.TiCDCChangefeedURIConfig,
	replicaConfig *ticdc_v1.ReplicaConfig,
	ticdcClusterEnt *entity.TiCDCCluster,
	changefeedEnt *entity.TiCDCChangefeed,
) bizerr.BizError {
	region := newRegionFromTiCDCCluster(ticdcClusterEnt)
	regionalReq := &ticdc_v1.UpdateChangefeedRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
		TidbClusterId:  fmt.Sprintf("%d", changefeedEnt.UpStreamClusterID.Uint64()),
		ChangefeedId:   changefeedEnt.KernelChangefeedID,
		Changefeed: &ticdc_v1.ChangefeedConfig{
			SinkUri:       uriConfig.GenerateURI(),
			ReplicaConfig: replicaConfig,
		},
	}
	return m.changefeedCli.UpdateChangefeed(ctx, region, regionalReq)
}

//nolint:dupl
func (m *changefeedModelRegionalImpl) PauseChangefeed(
	ctx context.Context, req *domain.PauseChangefeedReq,
) (*domain.PauseChangefeedResp, bizerr.BizError) {
	changefeedEnt, err := m.changefeedRepo.Load(ctx, req.ChangefeedID)
	if err != nil {
		return nil, err.Chain("load changefeed entity")
	}
	if changefeedEnt.KernelMetrics.FeedState == entity.ChangefeedKernelStatusStopped {
		log.GetLogger(ctx).Info("changefeed is already stopped",
			zap.Uint64("changefeed_id", req.ChangefeedID),
			zap.String("changefeed_status", string(changefeedEnt.KernelMetrics.FeedState)))
		return &domain.PauseChangefeedResp{}, nil
	}
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, changefeedEnt.TiCDCClusterID.Uint64())
	if err != nil {
		return nil, err.Chain("load ticdc cluster entity")
	}
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.doPauseChangefeed(ctx, changefeedEnt, ticdcClusterEnt)
	})
	if transErr != nil {
		return nil, convertTransactionError(transErr)
	}
	return &domain.PauseChangefeedResp{}, nil
}

func (m *changefeedModelRegionalImpl) doPauseChangefeed(
	ctx context.Context, changefeedEnt *entity.TiCDCChangefeed, ticdcClusterEnt *entity.TiCDCCluster,
) bizerr.BizError {
	changefeedEnt.KernelMetrics.FeedState = entity.ChangefeedKernelStatusStopped
	if err := m.changefeedRepo.Save(ctx, changefeedEnt); err != nil {
		return err.Chain("update changefeed entity")
	}
	if err := m.pauseChangefeedByRegional(ctx, changefeedEnt, ticdcClusterEnt); err != nil {
		return err.Chain("pause changefeed by regional")
	}
	return nil
}

func (m *changefeedModelRegionalImpl) pauseChangefeedByRegional(
	ctx context.Context, changefeedEnt *entity.TiCDCChangefeed, ticdcClusterEnt *entity.TiCDCCluster,
) bizerr.BizError {
	region := newRegionFromTiCDCCluster(ticdcClusterEnt)
	regionalReq := &ticdc_v1.PauseChangefeedRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
		TidbClusterId:  fmt.Sprintf("%d", changefeedEnt.UpStreamClusterID.Uint64()),
		ChangefeedId:   changefeedEnt.KernelChangefeedID,
	}
	return m.changefeedCli.PauseChangefeed(ctx, region, regionalReq)
}

//nolint:dupl
func (m *changefeedModelRegionalImpl) ResumeChangefeed(
	ctx context.Context, req *domain.ResumeChangefeedReq,
) (*domain.ResumeChangefeedResp, bizerr.BizError) {
	changefeedEnt, err := m.changefeedRepo.Load(ctx, req.ChangefeedID)
	if err != nil {
		return nil, err.Chain("load changefeed entity")
	}
	if changefeedEnt.KernelMetrics.FeedState == entity.ChangefeedKernelStatusNormal {
		log.GetLogger(ctx).Info("changefeed is already normal",
			zap.Uint64("changefeed_id", req.ChangefeedID),
			zap.String("changefeed_status", string(changefeedEnt.KernelMetrics.FeedState)))
		return &domain.ResumeChangefeedResp{}, nil
	}

	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, changefeedEnt.TiCDCClusterID.Uint64())
	if err != nil {
		return nil, err.Chain("load ticdc cluster entity")
	}
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.doResumeChangefeed(ctx, changefeedEnt, ticdcClusterEnt)
	})
	if transErr != nil {
		return nil, convertTransactionError(transErr)
	}
	return &domain.ResumeChangefeedResp{}, nil
}

func (m *changefeedModelRegionalImpl) doResumeChangefeed(
	ctx context.Context, changefeedEnt *entity.TiCDCChangefeed, ticdcClusterEnt *entity.TiCDCCluster,
) bizerr.BizError {
	changefeedEnt.KernelMetrics.FeedState = entity.ChangefeedKernelStatusNormal
	if err := m.changefeedRepo.Save(ctx, changefeedEnt); err != nil {
		return err.Chain("update changefeed entity")
	}
	if err := m.resumeChangefeedByRegional(ctx, changefeedEnt, ticdcClusterEnt); err != nil {
		return err.Chain("resume changefeed by regional")
	}
	return nil
}

func (m *changefeedModelRegionalImpl) resumeChangefeedByRegional(
	ctx context.Context, changefeedEnt *entity.TiCDCChangefeed, ticdcClusterEnt *entity.TiCDCCluster,
) bizerr.BizError {
	region := newRegionFromTiCDCCluster(ticdcClusterEnt)
	regionalReq := &ticdc_v1.ResumeChangefeedRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
		ChangefeedId:   changefeedEnt.KernelChangefeedID,
		TidbClusterId:  fmt.Sprintf("%d", changefeedEnt.UpStreamClusterID.Uint64()),
		// todo: support start from a specified checkpoint ts
		Config: &ticdc_v1.ResumeChangefeedConfig{
			OverwriteCheckpointTs: 0,
		},
	}
	return m.changefeedCli.ResumeChangefeed(ctx, region, regionalReq)
}

func (m *changefeedModelRegionalImpl) ListChangefeeds(
	ctx context.Context, req *domain.ListChangefeedsReq,
) (*domain.ListChangefeedsResp, bizerr.BizError) {
	changefeedEnts, err := m.changefeedRepo.List(ctx,
		m.preloadChangefeedOp(),
		m.listChangefeedsFilterOp(req.Filter),
		m.listChangefeedsPagerOp(req.Pager),
	)
	if err != nil {
		return nil, err.Chain("list changefeed entities")
	}
	count, err := m.changefeedRepo.Count(ctx, m.listChangefeedsFilterOp(req.Filter))
	if err != nil {
		return nil, err.Chain("count changefeed entities")
	}

	items := pie.Map(changefeedEnts, func(changefeedEnt entity.TiCDCChangefeed) domain.ListChangefeedItem {
		return domain.ListChangefeedItem{
			ChangefeedEntity: changefeedEnt,
		}
	})
	return &domain.ListChangefeedsResp{
		Items: items,
		Total: count,
	}, nil
}

func (m *changefeedModelRegionalImpl) QueryChangefeed(
	ctx context.Context, req *domain.QueryChangefeedReq,
) (*domain.QueryChangefeedResp, bizerr.BizError) {
	changefeedEnt, err := m.changefeedRepo.Get(ctx,
		m.preloadChangefeedOp(),
		m.listChangefeedsFilterOp(req.Filter),
	)
	if err != nil {
		return nil, err.Chain("query changefeed entity")
	}
	return &domain.QueryChangefeedResp{
		ChangefeedEntity: *changefeedEnt,
	}, nil
}

func (m *changefeedModelRegionalImpl) MarkChangefeedDeleted(
	ctx context.Context, req *domain.MarkChangefeedDeletedReq,
) (*domain.MarkChangefeedDeletedResp, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	changefeedEnt, err := m.changefeedRepo.Load(ctx, req.ChangefeedID)
	if err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			logger.Info("changefeed is not found", zap.Uint64("changefeed_id", req.ChangefeedID))
			return &domain.MarkChangefeedDeletedResp{}, nil
		}
		return nil, err.Chain("load changefeed entity")
	}
	if changefeedEnt.DeletedAt != nil {
		logger.Info("changefeed is already deleted",
			zap.Uint64("changefeed_id", req.ChangefeedID), zap.Time("deleted_at", *changefeedEnt.DeletedAt))
		return &domain.MarkChangefeedDeletedResp{}, nil
	}

	deletedAt := time.Now()
	changefeedEnt.DeletedAt = &deletedAt
	if err = m.changefeedRepo.Save(ctx, changefeedEnt); err != nil {
		return nil, err.Chain("save changefeed entity")
	}
	return &domain.MarkChangefeedDeletedResp{}, nil
}

func (m *changefeedModelRegionalImpl) GetChangefeedDetail(
	ctx context.Context, req *domain.GetChangefeedDetailReq,
) (*domain.GetChangefeedDetailResp, bizerr.BizError) {
	changefeedEnt, err := m.changefeedRepo.Load(ctx, req.ChangefeedID.Uint64())
	if err != nil {
		return nil, err.Chain("load changefeed entity")
	}
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, changefeedEnt.TiCDCClusterID.Uint64())
	if err != nil {
		return nil, err.Chain("load ticdc cluster entity")
	}

	region := newRegionFromTiCDCCluster(ticdcClusterEnt)
	regionalReq := &ticdc_v1.GetChangefeedRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
		TidbClusterId:  fmt.Sprintf("%d", changefeedEnt.UpStreamClusterID.Uint64()),
		ChangefeedId:   changefeedEnt.KernelChangefeedID,
	}
	cf, err := m.changefeedCli.GetChangefeed(ctx, region, regionalReq)
	if err != nil {
		return nil, err.Chain("query changefeed from regional")
	}
	changefeedPb, errPb := pbToStruct(cf)
	if errPb != nil {
		return nil, bizerr.WrapErrorWithCode(errPb, bizerr.ErrCodeUnknown, "convert to structpb.Struct")
	}
	return &domain.GetChangefeedDetailResp{
		Changefeed: changefeedPb,
	}, nil
}

type Value map[string]any

func (m *changefeedModelRegionalImpl) GetTiCDCCluster(
	ctx context.Context, req *domain.GetChangefeedClusterReq,
) (*domain.GetChangefeedClusterResp, bizerr.BizError) {
	changefeedEnt, err := m.changefeedRepo.Load(ctx, req.ChangefeedID.Uint64())
	if err != nil {
		return nil, err.Chain("load changefeed entity")
	}
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, changefeedEnt.TiCDCClusterID.Uint64())
	if err != nil {
		return nil, err.Chain("load ticdc cluster entity")
	}

	region := newRegionFromTiCDCCluster(ticdcClusterEnt)
	regionalReq := &ticdc_v1.GetTiCDCClusterRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
	}
	cluster, err := m.cdcClusterCli.GetTiCDCCluster(ctx, region, regionalReq)
	if err != nil {
		return nil, err.Chain("query changefeed from regional")
	}
	clusterPb, errPb := pbToStruct(cluster)
	if errPb != nil {
		return nil, bizerr.WrapErrorWithCode(errPb, bizerr.ErrCodeUnknown, "convert to structpb.Struct")
	}
	return &domain.GetChangefeedClusterResp{
		TiCDCCluster: clusterPb,
	}, nil
}

func (m *changefeedModelRegionalImpl) ListTiCDCClusters(
	ctx context.Context, req *domain.ListTiCDCClusterReq,
) (*domain.ListTiCDCClusterResp, bizerr.BizError) {
	region := &regionalapi.Region{
		SharedPoolID: req.Cluster.GetSharedPoolId(),
	}
	regionalReq := &ticdc_v1.ListTiCDCClustersRequest{
		TidbClusterId: fmt.Sprintf("%d", req.Cluster.GetClusterID()),
	}
	cluster, err := m.cdcClusterCli.ListTiCDCClusters(ctx, region, regionalReq)
	if err != nil {
		return nil, err.Chain("list ticdc cluster from regional")
	}
	clusterPb, errPb := pbToStruct(cluster)
	if errPb != nil {
		return nil, bizerr.WrapErrorWithCode(errPb, bizerr.ErrCodeUnknown, "convert to structpb.Struct")
	}
	return &domain.ListTiCDCClusterResp{
		TiCDCClusters: clusterPb,
	}, nil
}

func (m *changefeedModelRegionalImpl) preloadChangefeedOp() repo.TiCDCChangefeedFilterOp {
	return func(qs entity.TiCDCChangefeedQuerySet) entity.TiCDCChangefeedQuerySet {
		return qs.PreloadTiCDCCluster()
	}
}

func (m *changefeedModelRegionalImpl) listChangefeedsFilterOp(
	filter *domain.ListChangefeedFilter,
) repo.TiCDCChangefeedFilterOp {
	return func(qs entity.TiCDCChangefeedQuerySet) entity.TiCDCChangefeedQuerySet {
		if len(filter.TiCDCClusterIDs) > 0 {
			qs = qs.TiCDCClusterIDIn(pie.Map(filter.TiCDCClusterIDs,
				func(t uint64) entity.ID { return entity.ID(t) })...)
		}
		if len(filter.KernelChangefeedIDs) > 0 {
			qs = qs.IDIn(pie.Map(filter.KernelChangefeedIDs, func(t uint64) entity.ID { return entity.ID(t) })...)
		}
		if !filter.Deleted {
			qs = qs.DeletedAtIsNull()
		}
		if len(filter.BizChangefeedIDs) > 0 {
			qs = qs.BizChangefeedIDIn(pie.Map(filter.BizChangefeedIDs, func(t uint64) entity.ID { return entity.ID(t) })...)
		}
		return qs.OrderDescByCreatedAt()
	}
}

func (m *changefeedModelRegionalImpl) listChangefeedsPagerOp(pager *domain.Pager) repo.TiCDCChangefeedFilterOp {
	if pager == nil {
		pager = domain.NewPager()
	}
	return func(qs entity.TiCDCChangefeedQuerySet) entity.TiCDCChangefeedQuerySet {
		offset, limit := pager.CalculateOffsetAndLimit()
		qs = qs.Offset(int(offset)).Limit(int(limit))
		return qs
	}
}

func pbToStruct(pb proto.Message) (*structpb.Struct, error) {
	data, err := protojson.Marshal(pb)
	if err != nil {
		return nil, err
	}
	var mapValue map[string]any
	if err = json.Unmarshal(data, &mapValue); err != nil {
		return nil, err
	}
	structPb, errPb := structpb.NewStruct(mapValue)
	if errPb != nil {
		return nil, err
	}
	return structPb, nil
}