package service

import (
	"context"
	"strings"
	"testing"

	. "github.com/smartystreets/goconvey/convey"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/service/mock_service"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"go.uber.org/mock/gomock"
)

// identityEncryptor is a no-op EncryptorFacade used to drive ToChangefeed in
// tests without depending on a real KMS backend.
type identityEncryptor struct{}

func (identityEncryptor) EncryptBase64(ext string) (string, error) { return ext, nil }
func (identityEncryptor) DecryptBase64(encryptedEncoded string) (string, error) {
	return encryptedEncoded, nil
}

func newPrivateLinkKafkaInfo(connID, endpoints string, port int32) *entity.KafkaInfo {
	return &entity.KafkaInfo{
		Network: entity.KafkaNetwork{
			NetworkType:             constant.KafkaNetworkTypePrivateLink,
			PrivateLinkConnectionId: connID,
			PrivateLinkEndpoints:    endpoints,
			PrivateLinkPort:         port,
		},
		Broker: entity.Broker{
			KafkaVersion: constant.KafkaVersion3XX,
			Compression:  constant.KafkaCompressionTypeNone,
		},
		Authentication: entity.Authentication{
			AuthType:  constant.KafkaAuthTypeSASLScram512,
			UserName:  "tidbcdc",
			Password:  "secret",
			EnableTLS: true,
		},
		DataFormat: entity.DataFormat{Protocol: constant.CDCProtocolCanalJSON},
		TopicPartitionConfig: entity.TopicPartitionConfig{
			DefaultTopic:      "dev.tidbcdc.default",
			DispatchType:      constant.TopicDispatchTypeByTable,
			Separator:         ".",
			TopicPrefix:       "dev.tidbcdc.",
			PartitionNum:      6,
			ReplicationFactor: 3,
			PartitionDispatchers: []*entity.PartitionDispatcherConfig{
				{Matcher: []string{"*.*"}, PartitionType: constant.PartitionDispatcherTable},
			},
		},
	}
}

func newPrivateLinkMySQLInfo(connID, endpoint string, port int32) *entity.MySQLInfo {
	return &entity.MySQLInfo{
		Network: entity.MySQLNetwork{
			NetworkType:             constant.MySQLNetworkTypePrivateLink,
			PrivateLinkConnectionId: connID,
			PrivateLinkEndpoint:     endpoint,
			PrivateLinkPort:         port,
		},
		Authentication: entity.MySQLAuthentication{
			User:      "tidbcdc",
			Password:  "secret",
			EnableTLS: true,
		},
	}
}

func TestFailedChangefeedRecoveryStates(t *testing.T) {
	tests := []struct {
		name     string
		state    constant.ChangfeedState
		canPause bool
		canEdit  bool
	}{
		{name: "running", state: constant.ChangfeedStateRunning, canPause: true},
		{name: "paused", state: constant.ChangefeedStatePaused, canEdit: true},
		{name: "warning", state: constant.ChangefeedStateWarning, canPause: true},
		{name: "runtime failed", state: constant.ChangefeedStateFailed, canPause: true, canEdit: true},
		{name: "create failed", state: constant.ChangefeedStateCreateFailed},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := canPauseChangefeed(tt.state); got != tt.canPause {
				t.Fatalf("canPauseChangefeed(%q) = %v, want %v", tt.state, got, tt.canPause)
			}
			if got := canEditChangefeed(tt.state); got != tt.canEdit {
				t.Fatalf("canEditChangefeed(%q) = %v, want %v", tt.state, got, tt.canEdit)
			}
		})
	}
}

func TestSyncFailedKernelPreservesRequestedPausedState(t *testing.T) {
	changefeed := &entity.CDCChangefeed{
		ClusterID:     "cluster-1",
		ID:            "cf-1",
		State:         constant.ChangefeedStatePaused,
		CheckpointTSO: 100,
	}
	failedState := serverless.StateFailed
	checkpoint := uint64(123)

	SyncFromChangefeedCommon(context.Background(), changefeed, &serverless.ChangefeedCommonInfo{
		State:         failedState,
		CheckpointTSO: &checkpoint,
	})

	if changefeed.State != constant.ChangefeedStatePaused {
		t.Fatalf("state = %q, want %q", changefeed.State, constant.ChangefeedStatePaused)
	}
	if changefeed.CheckpointTSO != checkpoint {
		t.Fatalf("checkpoint = %d, want %d", changefeed.CheckpointTSO, checkpoint)
	}
}

// TestMergeKafkaNetworkFromOld covers the merge helper introduced for
// EditChangefeed private-link Kafka regressions. The console edit form only
// echoes the immutable identifiers (PrivateLinkConnectionId, NetworkType) and
// drops the resolved derivatives; this helper must carry them forward from
// the stored entity so the rebuilt SinkURI still has a host segment.
func TestMergeKafkaNetworkFromOld(t *testing.T) {
	const (
		oldConn      = "plc-old"
		oldEndpoints = "broker-1.example.com:9092,broker-2.example.com:9092"
		oldPort      = int32(9092)
	)

	Convey("Test_MergeKafkaNetworkFromOld", t, func() {
		Convey("private_link: empty fields are populated from the old entity", func() {
			old := newPrivateLinkKafkaInfo(oldConn, oldEndpoints, oldPort)
			// Simulate the VO that ToEntity produces from the console edit
			// payload: only ConnectionId and NetworkType are present.
			fresh := newPrivateLinkKafkaInfo(oldConn, "", 0)

			mergeKafkaNetworkFromOld(fresh, old)

			So(fresh.Network.PrivateLinkConnectionId, ShouldEqual, oldConn)
			So(fresh.Network.PrivateLinkEndpoints, ShouldEqual, oldEndpoints)
			So(fresh.Network.PrivateLinkPort, ShouldEqual, oldPort)
		})

		Convey("private_link: explicit non-empty values from the request are preserved", func() {
			old := newPrivateLinkKafkaInfo(oldConn, oldEndpoints, oldPort)
			newEndpoints := "broker-3.example.com:9093"
			fresh := newPrivateLinkKafkaInfo(oldConn, newEndpoints, 9093)

			mergeKafkaNetworkFromOld(fresh, old)

			So(fresh.Network.PrivateLinkEndpoints, ShouldEqual, newEndpoints)
			So(fresh.Network.PrivateLinkPort, ShouldEqual, int32(9093))
		})

		Convey("private_link: changing PrivateLinkConnectionId disables the carry-over", func() {
			// User pointed the changefeed at a different private link
			// connection. The stale endpoints from the old connection must
			// not bleed into the new one — caller is expected to re-resolve.
			old := newPrivateLinkKafkaInfo(oldConn, oldEndpoints, oldPort)
			fresh := newPrivateLinkKafkaInfo("plc-new", "", 0)

			mergeKafkaNetworkFromOld(fresh, old)

			So(fresh.Network.PrivateLinkConnectionId, ShouldEqual, "plc-new")
			So(fresh.Network.PrivateLinkEndpoints, ShouldEqual, "")
			So(fresh.Network.PrivateLinkPort, ShouldEqual, int32(0))
		})

		Convey("network type switch (public <-> private_link) disables the carry-over", func() {
			old := newPrivateLinkKafkaInfo(oldConn, oldEndpoints, oldPort)
			fresh := &entity.KafkaInfo{
				Network: entity.KafkaNetwork{NetworkType: constant.KafkaNetworkTypePublic},
			}

			mergeKafkaNetworkFromOld(fresh, old)

			// Public branch fills nothing from the old private-link record.
			So(fresh.Network.PublicEndpoints, ShouldEqual, "")
			So(fresh.Network.PrivateLinkEndpoints, ShouldEqual, "")
		})

		Convey("public: empty PublicEndpoints is populated from the old entity", func() {
			old := &entity.KafkaInfo{
				Network: entity.KafkaNetwork{
					NetworkType:     constant.KafkaNetworkTypePublic,
					PublicEndpoints: "public-broker.example.com:9092",
				},
			}
			fresh := &entity.KafkaInfo{
				Network: entity.KafkaNetwork{NetworkType: constant.KafkaNetworkTypePublic},
			}

			mergeKafkaNetworkFromOld(fresh, old)

			So(fresh.Network.PublicEndpoints, ShouldEqual, "public-broker.example.com:9092")
		})

		Convey("nil inputs are tolerated", func() {
			So(func() { mergeKafkaNetworkFromOld(nil, nil) }, ShouldNotPanic)
			old := newPrivateLinkKafkaInfo(oldConn, oldEndpoints, oldPort)
			So(func() { mergeKafkaNetworkFromOld(nil, old) }, ShouldNotPanic)
			fresh := newPrivateLinkKafkaInfo(oldConn, "", 0)
			So(func() { mergeKafkaNetworkFromOld(fresh, nil) }, ShouldNotPanic)
		})
	})
}

func TestMergeMySQLNetworkFromOld(t *testing.T) {
	const (
		oldConn     = "plc-mysql-old"
		oldEndpoint = "mysql.example.com:4000"
		oldPort     = int32(4000)
	)

	Convey("Test_MergeMySQLNetworkFromOld", t, func() {
		Convey("private_link: empty fields are populated from the old entity", func() {
			old := newPrivateLinkMySQLInfo(oldConn, oldEndpoint, oldPort)
			fresh := newPrivateLinkMySQLInfo(oldConn, "", 0)

			mergeMySQLNetworkFromOld(fresh, old)

			So(fresh.Network.PrivateLinkEndpoint, ShouldEqual, oldEndpoint)
			So(fresh.Network.PrivateLinkPort, ShouldEqual, oldPort)
		})

		Convey("private_link: changing PrivateLinkConnectionId disables the carry-over", func() {
			old := newPrivateLinkMySQLInfo(oldConn, oldEndpoint, oldPort)
			fresh := newPrivateLinkMySQLInfo("plc-mysql-new", "", 0)

			mergeMySQLNetworkFromOld(fresh, old)

			So(fresh.Network.PrivateLinkConnectionId, ShouldEqual, "plc-mysql-new")
			So(fresh.Network.PrivateLinkEndpoint, ShouldEqual, "")
		})

		Convey("public: empty PublicEndpoint is populated from the old entity", func() {
			old := &entity.MySQLInfo{
				Network: entity.MySQLNetwork{
					NetworkType:    constant.MySQLNetworkTypePublic,
					PublicEndpoint: "mysql-public.example.com:4000",
				},
			}
			fresh := &entity.MySQLInfo{
				Network: entity.MySQLNetwork{NetworkType: constant.MySQLNetworkTypePublic},
			}

			mergeMySQLNetworkFromOld(fresh, old)

			So(fresh.Network.PublicEndpoint, ShouldEqual, "mysql-public.example.com:4000")
		})

		Convey("nil inputs are tolerated", func() {
			So(func() { mergeMySQLNetworkFromOld(nil, nil) }, ShouldNotPanic)
		})
	})
}

// TestEditChangefeed_SinkURIPreservedAfterMerge is the end-to-end guard for
// the regression that motivated this fix: an Edit request that drops the
// resolved private-link endpoints must NOT cause ToChangefeed to emit a
// host-less SinkURI like "kafka:///topic", which would make TiCDC reject
// the update with ErrKafkaNewProducer "missing address".
func TestEditChangefeed_SinkURIPreservedAfterMerge(t *testing.T) {
	const (
		connID       = "plc-hlzbbiws35glbfes67gmr34twe"
		brokerList   = "boot-cvu.example.com:14098,boot-xcg.example.com:14099,boot-era.example.com:14100"
		defaultTopic = "dev.usw2.plaud.tidbcdc.default"
	)
	ctx := context.Background()
	enc := identityEncryptor{}

	Convey("Test_EditChangefeed_SinkURIPreservedAfterMerge", t, func() {
		Convey("kafka private_link: SinkURI still contains the broker list after merge", func() {
			// Old DB record carries the resolved broker list.
			oldEntity := &entity.CDCChangefeed{
				ID:       "cfd-test-id",
				SinkType: constant.ChangefeedTypeKafka,
				Kafka:    newPrivateLinkKafkaInfo(connID, brokerList, 0),
				Filter: &entity.CDCFilter{
					Mode:       constant.TableModeIgnoreNotSupportTable,
					FilterRule: []string{"db.t"},
				},
			}
			oldEntity.Kafka.TopicPartitionConfig.DefaultTopic = defaultTopic

			// New entity reconstructed from the console edit payload — note
			// the empty PrivateLinkEndpoints / Port. This mirrors what the
			// real production VO looked like in the on-call incident.
			newKafka := newPrivateLinkKafkaInfo(connID, "", 0)
			newKafka.TopicPartitionConfig.DefaultTopic = defaultTopic
			newFilter := &entity.CDCFilter{
				Mode:       constant.TableModeIgnoreNotSupportTable,
				FilterRule: []string{"db.t_new"},
			}

			// Apply the same merge that the fixed combineChangefeed does.
			mergeKafkaNetworkFromOld(newKafka, oldEntity.Kafka)
			oldEntity.Kafka = newKafka
			oldEntity.Filter = newFilter

			cfg, err := oldEntity.ToChangefeedForUpdate(ctx, enc)
			So(err, ShouldBeNil)
			So(cfg.SinkURI, ShouldNotBeNil)
			sinkURI := *cfg.SinkURI

			// Regression contract: the SinkURI MUST contain the broker hosts
			// — i.e., the segment between "kafka://" and the topic must be
			// non-empty and match the stored broker list.
			So(strings.HasPrefix(sinkURI, "kafka://"), ShouldBeTrue)
			So(sinkURI, ShouldNotEqual, "kafka:///"+defaultTopic)
			So(strings.Contains(sinkURI, brokerList), ShouldBeTrue)
			So(strings.HasSuffix(sinkURI, "/"+defaultTopic), ShouldBeTrue)
		})

		Convey("mysql private_link: SinkURI still contains the endpoint after merge", func() {
			const (
				mysqlConn = "plc-mysql"
				mysqlEP   = "mysql-priv.example.com:4000"
			)
			oldEntity := &entity.CDCChangefeed{
				SinkType: constant.ChangefeedTypeMySQL,
				MySQL:    newPrivateLinkMySQLInfo(mysqlConn, mysqlEP, 0),
				Filter: &entity.CDCFilter{
					Mode:       constant.TableModeIgnoreNotSupportTable,
					FilterRule: []string{"db.t"},
				},
			}
			newMySQL := newPrivateLinkMySQLInfo(mysqlConn, "", 0)
			newFilter := &entity.CDCFilter{
				Mode:       constant.TableModeIgnoreNotSupportTable,
				FilterRule: []string{"db.t_new"},
			}
			mergeMySQLNetworkFromOld(newMySQL, oldEntity.MySQL)
			oldEntity.MySQL = newMySQL
			oldEntity.Filter = newFilter

			cfg, err := oldEntity.ToChangefeedForUpdate(ctx, enc)
			So(err, ShouldBeNil)
			So(cfg.SinkURI, ShouldNotBeNil)
			sinkURI := *cfg.SinkURI
			So(strings.Contains(sinkURI, mysqlEP), ShouldBeTrue)
			// The endpoint must appear after the credentials separator (@)
			// and the URI must not collapse to "mysql://user:pass@".
			So(strings.HasSuffix(sinkURI, "@"), ShouldBeFalse)
		})
	})
}

// TestResolvePrivateLinkEndpoints exercises the explicit re-resolve step that
// runs in EditChangefeed when combineChangefeed could not carry the stored
// endpoints forward (e.g. the user switched PrivateLinkConnectionId).
func TestResolvePrivateLinkEndpoints(t *testing.T) {
	ctx := context.Background()

	Convey("Test_ResolvePrivateLinkEndpoints", t, func() {
		Convey("kafka private_link with empty endpoints resolves via dry-run", func() {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()
			plcSvc := mock_service.NewMockPrivateLinkConnectionService(ctrl)
			svc := &cdcService{privateLinkConnectionSvc: plcSvc}

			cf := &entity.CDCChangefeed{
				ID:        "cfd-test",
				ClusterID: "cluster-1",
				SinkType:  constant.ChangefeedTypeKafka,
				Kafka:     newPrivateLinkKafkaInfo("plc-new", "", 9092),
			}

			// dryrun=true, bizId="" — we only want the resolved domains.
			plcSvc.EXPECT().RegisterPrivateLinkConnection(
				ctx, "cluster-1", "plc-new", "", true,
			).Return([]string{"resolved-broker.example.com:9092"}, nil)

			err := svc.resolvePrivateLinkEndpoints(ctx, cf)
			So(err, ShouldBeNil)
			// buildPrivateLinkEndpoint keeps the resolved string verbatim when
			// it already contains ":" and an amazonaws.com suffix; otherwise
			// it appends the port. For non-MSK we expect host:port.
			So(cf.Kafka.Network.PrivateLinkEndpoints, ShouldContainSubstring, "resolved-broker.example.com")
		})

		Convey("kafka private_link with non-empty endpoints is a no-op", func() {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()
			plcSvc := mock_service.NewMockPrivateLinkConnectionService(ctrl)
			svc := &cdcService{privateLinkConnectionSvc: plcSvc}

			cf := &entity.CDCChangefeed{
				SinkType: constant.ChangefeedTypeKafka,
				Kafka:    newPrivateLinkKafkaInfo("plc", "broker.example.com:9092", 9092),
			}
			// No EXPECT — Finish() would fail if RegisterPrivateLinkConnection got called.
			err := svc.resolvePrivateLinkEndpoints(ctx, cf)
			So(err, ShouldBeNil)
			So(cf.Kafka.Network.PrivateLinkEndpoints, ShouldEqual, "broker.example.com:9092")
		})

		Convey("kafka public network is a no-op", func() {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()
			plcSvc := mock_service.NewMockPrivateLinkConnectionService(ctrl)
			svc := &cdcService{privateLinkConnectionSvc: plcSvc}

			cf := &entity.CDCChangefeed{
				SinkType: constant.ChangefeedTypeKafka,
				Kafka: &entity.KafkaInfo{
					Network: entity.KafkaNetwork{
						NetworkType:     constant.KafkaNetworkTypePublic,
						PublicEndpoints: "public.example.com:9092",
					},
				},
			}
			err := svc.resolvePrivateLinkEndpoints(ctx, cf)
			So(err, ShouldBeNil)
		})

		Convey("kafka private_link with ServerName skips re-resolution", func() {
			// ServerName takes precedence in ToChangefeed; PrivateLinkEndpoints
			// is not used for SinkURI in that branch.
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()
			plcSvc := mock_service.NewMockPrivateLinkConnectionService(ctrl)
			svc := &cdcService{privateLinkConnectionSvc: plcSvc}

			cf := &entity.CDCChangefeed{
				SinkType: constant.ChangefeedTypeKafka,
				Kafka:    newPrivateLinkKafkaInfo("plc", "", 9092),
			}
			cf.Kafka.Authentication.ServerName = "kafka.private.example.com"

			err := svc.resolvePrivateLinkEndpoints(ctx, cf)
			So(err, ShouldBeNil)
			So(cf.Kafka.Network.PrivateLinkEndpoints, ShouldEqual, "")
		})

		Convey("mysql private_link with empty endpoint resolves via dry-run", func() {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()
			plcSvc := mock_service.NewMockPrivateLinkConnectionService(ctrl)
			svc := &cdcService{privateLinkConnectionSvc: plcSvc}

			cf := &entity.CDCChangefeed{
				ID:        "cfd-mysql",
				ClusterID: "cluster-2",
				SinkType:  constant.ChangefeedTypeMySQL,
				MySQL:     newPrivateLinkMySQLInfo("plc-mysql-new", "", 4000),
			}

			plcSvc.EXPECT().RegisterPrivateLinkConnection(
				ctx, "cluster-2", "plc-mysql-new", "", true,
			).Return([]string{"resolved-mysql.example.com"}, nil)

			err := svc.resolvePrivateLinkEndpoints(ctx, cf)
			So(err, ShouldBeNil)
			So(cf.MySQL.Network.PrivateLinkEndpoint, ShouldEqual, "resolved-mysql.example.com:4000")
		})

		Convey("RegisterPrivateLinkConnection failure propagates", func() {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()
			plcSvc := mock_service.NewMockPrivateLinkConnectionService(ctrl)
			svc := &cdcService{privateLinkConnectionSvc: plcSvc}

			cf := &entity.CDCChangefeed{
				SinkType: constant.ChangefeedTypeKafka,
				Kafka:    newPrivateLinkKafkaInfo("plc-broken", "", 9092),
			}
			boom := bizerr.New(bizerr.ErrCodeUnknown, "boom")
			plcSvc.EXPECT().RegisterPrivateLinkConnection(
				ctx, gomock.Any(), "plc-broken", "", true,
			).Return(nil, boom)

			err := svc.resolvePrivateLinkEndpoints(ctx, cf)
			So(err, ShouldNotBeNil)
			So(cf.Kafka.Network.PrivateLinkEndpoints, ShouldEqual, "")
		})

		Convey("nil Kafka / MySQL pointers are tolerated", func() {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()
			plcSvc := mock_service.NewMockPrivateLinkConnectionService(ctrl)
			svc := &cdcService{privateLinkConnectionSvc: plcSvc}

			// SinkType says kafka but Kafka pointer is nil — guard must not panic.
			So(func() {
				_ = svc.resolvePrivateLinkEndpoints(ctx, &entity.CDCChangefeed{SinkType: constant.ChangefeedTypeKafka})
			}, ShouldNotPanic)

			So(func() {
				_ = svc.resolvePrivateLinkEndpoints(ctx, &entity.CDCChangefeed{SinkType: constant.ChangefeedTypeMySQL})
			}, ShouldNotPanic)
		})
	})
}

// TestValidateSinkAndNetworkUnchanged covers the EditChangefeed precheck that
// rejects sink-type / network-topology changes. This is the front-line guard
// against the Copilot-flagged "switch PLC during edit" hazards (stale
// PrivateLinkConnectionLink rows, ignored Port changes).
func TestValidateSinkAndNetworkUnchanged(t *testing.T) {
	const (
		oldConn      = "plc-old"
		newConn      = "plc-new"
		oldEndpoints = "broker-old.example.com:9092"
		oldPort      = int32(9092)
		newPort      = int32(9093)
	)

	mkKafkaVO := func(kafka *entity.KafkaInfo) *vo.Changefeed {
		return &vo.Changefeed{Sink: &vo.SinkInfo{Type: constant.ChangefeedTypeKafka, Kafka: kafka}}
	}
	mkMySQLVO := func(mysql *entity.MySQLInfo) *vo.Changefeed {
		return &vo.Changefeed{Sink: &vo.SinkInfo{Type: constant.ChangefeedTypeMySQL, MySQL: mysql}}
	}

	Convey("Test_ValidateSinkAndNetworkUnchanged", t, func() {
		Convey("sink type change is rejected", func() {
			old := &entity.CDCChangefeed{
				SinkType: constant.ChangefeedTypeKafka,
				Kafka:    newPrivateLinkKafkaInfo(oldConn, oldEndpoints, oldPort),
			}
			edited := &vo.Changefeed{Sink: &vo.SinkInfo{Type: constant.ChangefeedTypeMySQL}}

			err := validateSinkAndNetworkUnchanged(old, edited)
			So(err, ShouldNotBeNil)
			So(err.Error(), ShouldContainSubstring, "sink type")
		})

		Convey("kafka network type change is rejected", func() {
			old := &entity.CDCChangefeed{
				SinkType: constant.ChangefeedTypeKafka,
				Kafka:    newPrivateLinkKafkaInfo(oldConn, oldEndpoints, oldPort),
			}
			edited := mkKafkaVO(&entity.KafkaInfo{
				Network: entity.KafkaNetwork{NetworkType: constant.KafkaNetworkTypePublic},
			})
			err := validateSinkAndNetworkUnchanged(old, edited)
			So(err, ShouldNotBeNil)
			So(err.Error(), ShouldContainSubstring, "kafka network type")
		})

		Convey("kafka PrivateLinkConnectionId change is rejected", func() {
			old := &entity.CDCChangefeed{
				SinkType: constant.ChangefeedTypeKafka,
				Kafka:    newPrivateLinkKafkaInfo(oldConn, oldEndpoints, oldPort),
			}
			edited := mkKafkaVO(newPrivateLinkKafkaInfo(newConn, "", oldPort))

			err := validateSinkAndNetworkUnchanged(old, edited)
			So(err, ShouldNotBeNil)
			So(err.Error(), ShouldContainSubstring, "private link connection")
		})

		Convey("kafka PrivateLinkPort change is rejected", func() {
			old := &entity.CDCChangefeed{
				SinkType: constant.ChangefeedTypeKafka,
				Kafka:    newPrivateLinkKafkaInfo(oldConn, oldEndpoints, oldPort),
			}
			edited := mkKafkaVO(newPrivateLinkKafkaInfo(oldConn, "", newPort))

			err := validateSinkAndNetworkUnchanged(old, edited)
			So(err, ShouldNotBeNil)
			So(err.Error(), ShouldContainSubstring, "private link port")
		})

		Convey("kafka edit with same network identifiers passes (empty fields tolerated)", func() {
			old := &entity.CDCChangefeed{
				SinkType: constant.ChangefeedTypeKafka,
				Kafka:    newPrivateLinkKafkaInfo(oldConn, oldEndpoints, oldPort),
			}
			// Console edit payload typically omits derived fields entirely.
			edited := mkKafkaVO(&entity.KafkaInfo{
				Network: entity.KafkaNetwork{
					NetworkType:             constant.KafkaNetworkTypePrivateLink,
					PrivateLinkConnectionId: oldConn,
				},
			})
			So(validateSinkAndNetworkUnchanged(old, edited), ShouldBeNil)
		})

		Convey("kafka edit with matching ConnectionId and Port passes", func() {
			old := &entity.CDCChangefeed{
				SinkType: constant.ChangefeedTypeKafka,
				Kafka:    newPrivateLinkKafkaInfo(oldConn, oldEndpoints, oldPort),
			}
			edited := mkKafkaVO(newPrivateLinkKafkaInfo(oldConn, "", oldPort))
			So(validateSinkAndNetworkUnchanged(old, edited), ShouldBeNil)
		})

		Convey("mysql PrivateLinkConnectionId change is rejected", func() {
			old := &entity.CDCChangefeed{
				SinkType: constant.ChangefeedTypeMySQL,
				MySQL:    newPrivateLinkMySQLInfo(oldConn, "mysql.example.com:4000", 4000),
			}
			edited := mkMySQLVO(newPrivateLinkMySQLInfo(newConn, "", 4000))
			err := validateSinkAndNetworkUnchanged(old, edited)
			So(err, ShouldNotBeNil)
			So(err.Error(), ShouldContainSubstring, "private link connection")
		})

		Convey("mysql PrivateLinkPort change is rejected", func() {
			old := &entity.CDCChangefeed{
				SinkType: constant.ChangefeedTypeMySQL,
				MySQL:    newPrivateLinkMySQLInfo(oldConn, "mysql.example.com:4000", 4000),
			}
			edited := mkMySQLVO(newPrivateLinkMySQLInfo(oldConn, "", 4001))
			err := validateSinkAndNetworkUnchanged(old, edited)
			So(err, ShouldNotBeNil)
			So(err.Error(), ShouldContainSubstring, "private link port")
		})

		Convey("public kafka edit allows changing PublicEndpoints", func() {
			// PublicEndpoints is user-supplied configuration, not a derived
			// identifier — tweaking it during edit is legitimate.
			old := &entity.CDCChangefeed{
				SinkType: constant.ChangefeedTypeKafka,
				Kafka: &entity.KafkaInfo{
					Network: entity.KafkaNetwork{
						NetworkType:     constant.KafkaNetworkTypePublic,
						PublicEndpoints: "old.example.com:9092",
					},
				},
			}
			edited := mkKafkaVO(&entity.KafkaInfo{
				Network: entity.KafkaNetwork{
					NetworkType:     constant.KafkaNetworkTypePublic,
					PublicEndpoints: "new.example.com:9092",
				},
			})
			So(validateSinkAndNetworkUnchanged(old, edited), ShouldBeNil)
		})
	})
}