package changefeed

import (
	"context"
	"strconv"
	"testing"
	"time"

	"github.com/bufbuild/protovalidate-go"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service-ng/app/repo/mock"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	changefeedopenpb "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/openapi/v1beta2"
	"github.com/tidbcloud/pkgv2/bizerr"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/structpb"
)

type consoleChangefeedStub struct {
	createChangefeed    func(context.Context, *dataflow.ChangefeedCreateReq) (*dataflow.ChangefeedID, error)
	testChangefeed      func(context.Context, *dataflow.ChangefeedCreateReq) (*emptypb.Empty, error)
	editChangefeed      func(context.Context, *dataflow.ChangefeedEditReq) (*emptypb.Empty, error)
	deleteChangefeed    func(context.Context, *dataflow.ChangefeedID) (*emptypb.Empty, error)
	getChangefeed       func(context.Context, *dataflow.ChangefeedID) (*dataflow.Changefeed, error)
	getChangefeedDetail func(context.Context, *dataflow.ChangefeedID) (*structpb.Struct, error)
}

func (s *consoleChangefeedStub) ListChangefeeds(
	context.Context, *dataflow.ChangefeedListReq,
) (*dataflow.Changefeeds, error) {
	panic("unexpected ListChangefeeds call")
}

func (s *consoleChangefeedStub) GetChangefeed(
	ctx context.Context, req *dataflow.ChangefeedID,
) (*dataflow.Changefeed, error) {
	if s.getChangefeed == nil {
		panic("unexpected GetChangefeed call")
	}
	return s.getChangefeed(ctx, req)
}

func (s *consoleChangefeedStub) GetChangefeedDetail(
	ctx context.Context, req *dataflow.ChangefeedID,
) (*structpb.Struct, error) {
	if s.getChangefeedDetail == nil {
		panic("unexpected GetChangefeedDetail call")
	}
	return s.getChangefeedDetail(ctx, req)
}

func (s *consoleChangefeedStub) ScaleChangefeed(
	context.Context, *dataflow.ChangefeedScaleReq,
) (*emptypb.Empty, error) {
	panic("unexpected ScaleChangefeed call")
}

func (s *consoleChangefeedStub) ResumeChangefeed(
	context.Context, *dataflow.ChangefeedID,
) (*emptypb.Empty, error) {
	panic("unexpected ResumeChangefeed call")
}

func (s *consoleChangefeedStub) PauseChangefeed(
	context.Context, *dataflow.ChangefeedID,
) (*emptypb.Empty, error) {
	panic("unexpected PauseChangefeed call")
}

func (s *consoleChangefeedStub) CreateChangefeed(
	ctx context.Context, req *dataflow.ChangefeedCreateReq,
) (*dataflow.ChangefeedID, error) {
	if s.createChangefeed == nil {
		panic("unexpected CreateChangefeed call")
	}
	return s.createChangefeed(ctx, req)
}

func (s *consoleChangefeedStub) EditChangefeed(
	ctx context.Context, req *dataflow.ChangefeedEditReq,
) (*emptypb.Empty, error) {
	if s.editChangefeed == nil {
		panic("unexpected EditChangefeed call")
	}
	return s.editChangefeed(ctx, req)
}

func (s *consoleChangefeedStub) DeleteChangefeed(
	ctx context.Context, req *dataflow.ChangefeedID,
) (*emptypb.Empty, error) {
	if s.deleteChangefeed == nil {
		panic("unexpected DeleteChangefeed call")
	}
	return s.deleteChangefeed(ctx, req)
}

func (s *consoleChangefeedStub) TestChangefeed(
	ctx context.Context, req *dataflow.ChangefeedCreateReq,
) (*emptypb.Empty, error) {
	return s.testChangefeed(ctx, req)
}

type consoleSpecStub struct{}

func (*consoleSpecStub) List(
	context.Context, *dataflow.ChangefeedSpecListReq,
) (*dataflow.ChangefeedSpecs, error) {
	return &dataflow.ChangefeedSpecs{}, nil
}

func TestCreateChangefeedDryRunUsesConsoleCreate(t *testing.T) {
	createCalls := 0
	consoleAPI := &consoleChangefeedStub{
		createChangefeed: func(_ context.Context, req *dataflow.ChangefeedCreateReq) (*dataflow.ChangefeedID, error) {
			createCalls++
			require.Equal(t, uint64(12345), req.GetClusterId())
			require.Equal(t, dataflow.ChangefeedType_MYSQL, req.GetChangefeedType())
			require.Equal(t, "target.example.com:4000", req.GetMysqlInfo().GetConnection().GetEndpoint())
			require.True(t, req.GetDryRun())
			return &dataflow.ChangefeedID{}, nil
		},
	}
	adapter := newConsoleChangefeedAdapter(consoleAPI, &consoleSpecStub{}, nil)

	got, err := adapter.CreateChangefeed(context.Background(), &changefeedopenpb.CreateChangefeedRequest{
		DryRun: true,
		Changefeed: &changefeedopenpb.Changefeed{
			TidbId:         "12345",
			DownstreamType: changefeedopenpb.DownstreamType_MYSQL,
			DownstreamConfig: &changefeedopenpb.Changefeed_Mysql{
				Mysql: &changefeedopenpb.MySQLDownstreamConfig{
					Connection: &changefeedopenpb.Connection{
						Endpoint: "target.example.com:4000",
						Username: "root",
					},
				},
			},
		},
	}, 12345)

	require.NoError(t, err)
	require.Equal(t, 1, createCalls)
	require.Equal(t, &changefeedopenpb.Changefeed{}, got)
}

func TestEditChangefeedDryRunUsesConsoleEdit(t *testing.T) {
	changefeedRepo := repo_mock.NewBizChangefeedRepo(t)
	changefeedRepo.On("Load", mock.Anything, uint64(67890)).Return(&entity.BizChangefeed{
		ID:        entity.ID(67890),
		ClusterID: entity.ID(12345),
	}, nil)

	testCalls := 0
	consoleAPI := &consoleChangefeedStub{
		getChangefeed: func(_ context.Context, req *dataflow.ChangefeedID) (*dataflow.Changefeed, error) {
			require.Equal(t, uint64(67890), req.GetChangefeedId())
			return &dataflow.Changefeed{
				ChangefeedId:   67890,
				ChangefeedType: dataflow.ChangefeedType_MYSQL,
				Request: &dataflow.ChangefeedCreateReq{
					ClusterId: 12345,
					MysqlInfo: &dataflow.MySQLInfo{Connection: &dataflow.Connection{
						Endpoint: "target.example.com:4000",
						Username: "root",
					}},
				},
			}, nil
		},
		editChangefeed: func(_ context.Context, req *dataflow.ChangefeedEditReq) (*emptypb.Empty, error) {
			testCalls++
			require.Equal(t, uint64(67890), req.GetChangefeedId())
			require.Equal(t, uint64(12345), req.GetClusterId())
			require.True(t, req.GetDryRun())
			require.Empty(t, req.GetMysqlInfo().GetConnection().GetPassword())
			return &emptypb.Empty{}, nil
		},
	}
	adapter := newConsoleChangefeedAdapter(consoleAPI, &consoleSpecStub{}, changefeedRepo)

	got, err := adapter.EditChangefeedDownstreamConfig(context.Background(),
		&changefeedopenpb.EditChangefeedDownstreamConfigRequest{
			ChangefeedId:   "67890",
			DownstreamType: changefeedopenpb.DownstreamType_MYSQL,
			DryRun:         true,
		})

	require.NoError(t, err)
	require.Equal(t, 1, testCalls)
	require.Equal(t, &changefeedopenpb.Changefeed{}, got)
}

func TestDeleteChangefeedResolvesTiDBIDFromRepo(t *testing.T) {
	changefeedRepo := repo_mock.NewBizChangefeedRepo(t)
	changefeedRepo.On("Load", mock.Anything, uint64(67890)).Return(&entity.BizChangefeed{
		ID:        entity.ID(67890),
		ClusterID: entity.ID(12345),
	}, nil)

	consoleAPI := &consoleChangefeedStub{
		deleteChangefeed: func(_ context.Context, req *dataflow.ChangefeedID) (*emptypb.Empty, error) {
			require.Equal(t, uint64(67890), req.GetChangefeedId())
			require.Equal(t, uint64(12345), req.GetClusterId())
			return &emptypb.Empty{}, nil
		},
	}
	adapter := newConsoleChangefeedAdapter(consoleAPI, &consoleSpecStub{}, changefeedRepo)

	got, err := adapter.DeleteChangefeed(context.Background(), &changefeedopenpb.DeleteChangefeedRequest{
		ChangefeedId: "67890",
	})

	require.NoError(t, err)
	require.Equal(t, &emptypb.Empty{}, got)
}

func TestPremiumDownstreamConversions(t *testing.T) {
	t.Run("S3 compatible provider and BDR", func(t *testing.T) {
		bdrMode := true
		consoleChangefeed := &dataflow.Changefeed{
			ChangefeedId:   1,
			ChangefeedType: dataflow.ChangefeedType_S3,
			Status:         dataflow.Changefeed_RESUMING,
			ChangefeedSpec: "8rcu",
			Request: &dataflow.ChangefeedCreateReq{
				ClusterId: 10,
				S3Info: &dataflow.S3Info{
					Endpoint: &dataflow.S3Endpoint{
						S3Uri:    "s3://bucket/prefix",
						Endpoint: "https://s3.oss-us-east-1.aliyuncs.com",
						Provider: dataflow.S3Provider_S3_PROVIDER_ALIBABA_OSS,
					},
					TableConfig: &dataflow.TableConfig{BdrMode: &bdrMode},
				},
			},
		}

		got := convertConsoleToPublicChangefeed(consoleChangefeed)
		require.Equal(t, changefeedopenpb.Changefeed_RESUMING, got.GetState())
		require.Equal(t, "8ccu", got.GetChangefeedCapacity())
		require.True(t, got.GetTableConfig().GetBdrMode())
		require.Equal(t, "https://s3.oss-us-east-1.aliyuncs.com", got.GetS3().GetEndpoint().GetCustomEndpoint())
		require.Equal(t, changefeedopenpb.S3DownstreamConfig_Endpoint_ALIBABA_OSS, got.GetS3().GetEndpoint().GetProvider())
	})

	t.Run("Pulsar round trip", func(t *testing.T) {
		public := &changefeedopenpb.Changefeed{
			DownstreamConfig: &changefeedopenpb.Changefeed_Pulsar{
				Pulsar: &changefeedopenpb.PulsarDownstreamConfig{
					DestinationProtocol: changefeedopenpb.PulsarDownstreamConfig_PULSAR_SSL,
					Endpoint:            "pulsar.example.com:6651",
					Authentication: &changefeedopenpb.PulsarDownstreamConfig_Authentication{
						AuthType: changefeedopenpb.PulsarDownstreamConfig_Authentication_TOKEN,
						Token:    "secret",
					},
					Compression: changefeedopenpb.PulsarDownstreamConfig_ZSTD,
					DataFormat: &changefeedopenpb.PulsarDownstreamConfig_DataFormat{
						Protocol: changefeedopenpb.PulsarDownstreamConfig_DataFormat_CANAL_JSON,
					},
					TopicDistribution: &changefeedopenpb.PulsarDownstreamConfig_TopicDistribution{
						Tenant:       "tenant",
						Namespace:    "namespace",
						DispatchType: changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_BY_TABLE,
					},
				},
			},
		}

		internal := toDataflowPulsarInfo(public)
		require.Equal(t, dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR_SSL, internal.GetDestinationProtocol())
		require.Equal(t, "secret", internal.GetAuthentication().GetToken())
		require.Equal(t, dataflow.PulsarCompressionType_PULSAR_COMPRESSION_ZSTD, internal.GetCompressionType())

		got := toOpenAPIPulsarDownstreamConfig(internal)
		require.Equal(t, changefeedopenpb.PulsarDownstreamConfig_PULSAR_SSL, got.GetDestinationProtocol())
		require.Equal(t, changefeedopenpb.PulsarDownstreamConfig_Authentication_TOKEN, got.GetAuthentication().GetAuthType())
		require.Empty(t, got.GetAuthentication().GetToken(), "response must redact the token")
	})
}

func TestPremiumRequestValuesAreNotDefaulted(t *testing.T) {
	t.Run("table mode preserves explicit values", func(t *testing.T) {
		require.Equal(t, dataflow.TableMode_FORCE_SYNC, toDataflowTableConfig(&changefeedopenpb.TableConfig{
			Mode: changefeedopenpb.TableConfig_FORCE_SYNC,
		}).GetMode())
		require.Equal(t, dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE, toDataflowTableConfig(&changefeedopenpb.TableConfig{
			Mode: changefeedopenpb.TableConfig_IGNORE_NOT_SUPPORT_TABLE,
		}).GetMode())
	})

	t.Run("Kafka unspecified version is not rewritten to 3XX", func(t *testing.T) {
		require.Equal(t, dataflow.KafkaVersion_KAFKA_VERSION_0XX,
			toDataflowKafkaVersion(changefeedopenpb.KafkaDownstreamConfig_KAFKA_VERSION_UNSPECIFIED))
	})

	t.Run("storage and CSV values pass through unchanged", func(t *testing.T) {
		format := toDataflowS3DataFormat(&changefeedopenpb.S3DataFormat{
			Protocol:             changefeedopenpb.S3DataFormat_PROTOCOL_CSV,
			OutputRawChangeEvent: true,
			OutputFieldHeader:    true,
		})
		require.Empty(t, format.GetDelimiter())
		require.True(t, format.GetOutputRawChangeEvent())
		require.True(t, format.GetOutputFieldHeader())

		publicFormat := toOpenAPIS3DataFormat(format)
		require.True(t, publicFormat.GetOutputRawChangeEvent())
		require.True(t, publicFormat.GetOutputFieldHeader())

		flush := toDataflowStorageFlushConfig(&changefeedopenpb.StorageFlushConfig{})
		require.Empty(t, flush.GetFlushInterval())
		require.Zero(t, flush.GetFileSize())
	})

	t.Run("Pulsar values are not materialized from UI defaults", func(t *testing.T) {
		info := toDataflowPulsarConfig(&changefeedopenpb.PulsarDownstreamConfig{})
		require.Nil(t, info.GetAdvancedSetting())
		require.Empty(t, info.GetPartitionDispatchers())
	})
}

func TestEditDownstreamConfigProtovalidate(t *testing.T) {
	validator, err := protovalidate.New()
	require.NoError(t, err)

	t.Run("nil downstream config is valid", func(t *testing.T) {
		req := &changefeedopenpb.EditChangefeedDownstreamConfigRequest{
			ChangefeedId:   "1",
			DownstreamType: changefeedopenpb.DownstreamType_KAFKA,
		}
		require.NoError(t, validator.Validate(req))
	})

	t.Run("selected downstream config validates nested required fields", func(t *testing.T) {
		req := &changefeedopenpb.EditChangefeedDownstreamConfigRequest{
			ChangefeedId:   "1",
			DownstreamType: changefeedopenpb.DownstreamType_KAFKA,
			DownstreamConfig: &changefeedopenpb.EditChangefeedDownstreamConfigRequest_Kafka{
				Kafka: &changefeedopenpb.KafkaDownstreamConfig{},
			},
		}
		require.Error(t, validator.Validate(req))

		req.DownstreamConfig = &changefeedopenpb.EditChangefeedDownstreamConfigRequest_Kafka{
			Kafka: &changefeedopenpb.KafkaDownstreamConfig{
				Broker: &changefeedopenpb.KafkaDownstreamConfig_Broker{
					Version: changefeedopenpb.KafkaDownstreamConfig_KAFKA_VERSION_3XX,
				},
				Authentication: &changefeedopenpb.KafkaDownstreamConfig_Authentication{
					AuthType: changefeedopenpb.KafkaDownstreamConfig_Authentication_DISABLE,
				},
				DataFormat: &changefeedopenpb.KafkaDownstreamConfig_DataFormat{
					Protocol: changefeedopenpb.KafkaDownstreamConfig_DataFormat_PROTOCOL_CANAL_JSON,
				},
				TopicPartitionConfig: &changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig{
					DispatchType:      changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_BY_TABLE,
					ReplicationFactor: 3,
					PartitionNum:      6,
				},
			},
		}
		require.NoError(t, validator.Validate(req))
	})

	t.Run("Kafka edit may preserve passwords", func(t *testing.T) {
		kafka := validKafkaDownstreamConfig()
		kafka.Authentication.AuthType = changefeedopenpb.KafkaDownstreamConfig_Authentication_SASL_PLAIN
		kafka.Authentication.Username = stringPtr("kafka-user")
		kafka.DataFormat = validKafkaAvroDataFormat()
		registry := kafka.DataFormat.GetAvroConfig().GetSchemaRegistry()
		registry.EnableBasicAuth = true
		registry.BasicAuthUsername = "schema-user"

		req := &changefeedopenpb.EditChangefeedDownstreamConfigRequest{
			ChangefeedId:   "1",
			DownstreamType: changefeedopenpb.DownstreamType_KAFKA,
			DownstreamConfig: &changefeedopenpb.EditChangefeedDownstreamConfigRequest_Kafka{
				Kafka: kafka,
			},
		}
		require.NoError(t, validator.Validate(req))
	})

	t.Run("S3 storage flush config is optional", func(t *testing.T) {
		req := &changefeedopenpb.EditChangefeedDownstreamConfigRequest{
			ChangefeedId:   "1",
			DownstreamType: changefeedopenpb.DownstreamType_S3,
			DownstreamConfig: &changefeedopenpb.EditChangefeedDownstreamConfigRequest_S3{
				S3: &changefeedopenpb.S3DownstreamConfig{
					Endpoint: &changefeedopenpb.S3DownstreamConfig_Endpoint{
						S3Uri:     "s3://bucket/prefix",
						AuthType:  changefeedopenpb.S3DownstreamConfig_Endpoint_S3_AUTH_TYPE_AK_SK,
						AccessKey: "access-key",
					},
					DataFormat: &changefeedopenpb.S3DataFormat{
						Protocol:  changefeedopenpb.S3DataFormat_PROTOCOL_CANAL_JSON,
						Delimiter: ",",
					},
				},
			},
		}
		require.NoError(t, validator.Validate(req))
	})

	t.Run("Pulsar UI defaults are required API inputs", func(t *testing.T) {
		req := &changefeedopenpb.EditChangefeedDownstreamConfigRequest{
			ChangefeedId:   "1",
			DownstreamType: changefeedopenpb.DownstreamType_PULSAR,
			DownstreamConfig: &changefeedopenpb.EditChangefeedDownstreamConfigRequest_Pulsar{
				Pulsar: &changefeedopenpb.PulsarDownstreamConfig{
					DestinationProtocol: changefeedopenpb.PulsarDownstreamConfig_PULSAR,
					Endpoint:            "pulsar.example.com:6650",
					Authentication: &changefeedopenpb.PulsarDownstreamConfig_Authentication{
						AuthType: changefeedopenpb.PulsarDownstreamConfig_Authentication_BASIC,
						Username: "pulsar-user",
					},
					DataFormat: &changefeedopenpb.PulsarDownstreamConfig_DataFormat{
						Protocol: changefeedopenpb.PulsarDownstreamConfig_DataFormat_CANAL_JSON,
					},
					TopicDistribution: &changefeedopenpb.PulsarDownstreamConfig_TopicDistribution{
						Tenant:       "public",
						Namespace:    "default",
						DispatchType: changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_BY_TABLE,
					},
				},
			},
		}
		require.Error(t, validator.Validate(req))

		pulsar := req.GetPulsar()
		pulsar.PartitionDispatchers = []*changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig_PartitionDispatcherConfig{{
			PartitionType: changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig_TABLE,
			Matcher:       []string{"*.*"},
		}}
		pulsar.AdvancedSetting = &changefeedopenpb.PulsarDownstreamConfig_AdvancedSetting{
			BatchingMaxMessages:     1000,
			BatchingMaxPublishDelay: 10,
			ConnectionTimeout:       5,
			OperationTimeout:        30,
			SendTimeout:             30,
		}
		require.NoError(t, validator.Validate(req))
	})

	t.Run("mismatched downstream config is invalid", func(t *testing.T) {
		req := &changefeedopenpb.EditChangefeedDownstreamConfigRequest{
			ChangefeedId:   "1",
			DownstreamType: changefeedopenpb.DownstreamType_KAFKA,
			DownstreamConfig: &changefeedopenpb.EditChangefeedDownstreamConfigRequest_Mysql{
				Mysql: validMySQLDownstreamConfig(),
			},
		}
		require.Error(t, validator.Validate(req))
	})
}

func TestS3DataFormatDelimiterProtovalidate(t *testing.T) {
	validator, err := protovalidate.New()
	require.NoError(t, err)

	require.Error(t, validator.Validate(&changefeedopenpb.S3DataFormat{
		Protocol: changefeedopenpb.S3DataFormat_PROTOCOL_CSV,
	}))
	require.NoError(t, validator.Validate(&changefeedopenpb.S3DataFormat{
		Protocol:  changefeedopenpb.S3DataFormat_PROTOCOL_CSV,
		Delimiter: ",",
	}))
	require.NoError(t, validator.Validate(&changefeedopenpb.S3DataFormat{
		Protocol: changefeedopenpb.S3DataFormat_PROTOCOL_CANAL_JSON,
	}))
	require.Error(t, validator.Validate(&changefeedopenpb.S3DataFormat{
		Protocol: changefeedopenpb.S3DataFormat_Protocol(3),
	}))
}

func TestServerlessDownstreamTypeIsRejected(t *testing.T) {
	validator, err := protovalidate.New()
	require.NoError(t, err)
	serverless := changefeedopenpb.DownstreamType(4)

	require.Error(t, validator.Validate(&changefeedopenpb.ListChangefeedsRequest{
		TidbId:         "1",
		DownstreamType: &serverless,
	}))
	_, err = consoleTypeFromDownstreamType(serverless, true)
	require.Error(t, err)
	require.Equal(t, changefeedopenpb.DownstreamType_DOWNSTREAM_TYPE_UNSPECIFIED,
		downstreamTypeFromConsoleType(dataflow.ChangefeedType_SERVERLESS_TIDB))
}

func TestConditionalRequiredMessageProtovalidate(t *testing.T) {
	validator, err := protovalidate.New()
	require.NoError(t, err)

	t.Run("start position", func(t *testing.T) {
		require.NoError(t, validator.Validate(&changefeedopenpb.StartPosition{
			Mode: changefeedopenpb.StartPosition_FROM_NOW,
		}))
		require.Error(t, validator.Validate(&changefeedopenpb.StartPosition{
			Mode: changefeedopenpb.StartPosition_FROM_TSO,
		}))
		require.NoError(t, validator.Validate(&changefeedopenpb.StartPosition{
			Mode:     changefeedopenpb.StartPosition_FROM_TSO,
			Position: &changefeedopenpb.StartPosition_StartTso{StartTso: 1},
		}))
		require.Error(t, validator.Validate(&changefeedopenpb.StartPosition{
			Mode: changefeedopenpb.StartPosition_FROM_UTC,
		}))
		require.NoError(t, validator.Validate(&changefeedopenpb.StartPosition{
			Mode:     changefeedopenpb.StartPosition_FROM_UTC,
			Position: &changefeedopenpb.StartPosition_StartTimestamp{StartTimestamp: "2026-01-01T00:00:00Z"},
		}))
	})

	t.Run("network sink endpoint", func(t *testing.T) {
		require.NoError(t, validator.Validate(&changefeedopenpb.NetworkInfo{
			NetworkType: changefeedopenpb.NetworkInfo_NETWORK_TYPE_PUBLIC,
		}))
		require.NoError(t, validator.Validate(&changefeedopenpb.NetworkInfo{
			NetworkType: changefeedopenpb.NetworkInfo_NETWORK_TYPE_VPC_PEERING,
		}))
		require.Error(t, validator.Validate(&changefeedopenpb.NetworkInfo{
			NetworkType: changefeedopenpb.NetworkInfo_NETWORK_TYPE_PRIVATE_LINK,
		}))
		require.NoError(t, validator.Validate(&changefeedopenpb.NetworkInfo{
			NetworkType:    changefeedopenpb.NetworkInfo_NETWORK_TYPE_PRIVATE_LINK,
			SinkEndpointId: uint64Ptr(1),
		}))
	})

	t.Run("Kafka authentication", func(t *testing.T) {
		require.NoError(t, validator.Validate(&changefeedopenpb.KafkaDownstreamConfig_Authentication{
			AuthType: changefeedopenpb.KafkaDownstreamConfig_Authentication_DISABLE,
		}))
		require.Error(t, validator.Validate(&changefeedopenpb.KafkaDownstreamConfig_Authentication{
			AuthType: changefeedopenpb.KafkaDownstreamConfig_Authentication_SASL_PLAIN,
		}))
		require.NoError(t, validator.Validate(&changefeedopenpb.KafkaDownstreamConfig_Authentication{
			AuthType: changefeedopenpb.KafkaDownstreamConfig_Authentication_SASL_PLAIN,
			Username: stringPtr("kafka-user"),
		}))
	})

	t.Run("Kafka Avro configuration", func(t *testing.T) {
		require.Error(t, validator.Validate(&changefeedopenpb.KafkaDownstreamConfig_DataFormat{
			Protocol: changefeedopenpb.KafkaDownstreamConfig_DataFormat_PROTOCOL_AVRO,
		}))
		require.Error(t, validator.Validate(&changefeedopenpb.KafkaDownstreamConfig_DataFormat_AvroConfig{}))
		require.Error(t, validator.Validate(
			&changefeedopenpb.KafkaDownstreamConfig_DataFormat_AvroConfig_SchemaRegistry{}))
		require.Error(t, validator.Validate(
			&changefeedopenpb.KafkaDownstreamConfig_DataFormat_AvroConfig_SchemaRegistry{
				SchemaRegistryEndpoints: "https://registry.example.com",
				EnableBasicAuth:         true,
			}))
		require.NoError(t, validator.Validate(
			&changefeedopenpb.KafkaDownstreamConfig_DataFormat_AvroConfig_SchemaRegistry{
				SchemaRegistryEndpoints: "https://registry.example.com",
				EnableBasicAuth:         true,
				BasicAuthUsername:       "schema-user",
			}))

		kafka := validKafkaDownstreamConfig()
		kafka.DataFormat = validKafkaAvroDataFormat()
		kafka.TopicPartitionConfig.DispatchType =
			changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_ONE_TOPIC
		kafka.TopicPartitionConfig.DefaultTopic = "events"
		require.Error(t, validator.Validate(kafka))
		kafka.TopicPartitionConfig.DispatchType =
			changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_BY_TABLE
		kafka.TopicPartitionConfig.DefaultTopic = ""
		require.NoError(t, validator.Validate(kafka))
	})

	t.Run("topic and partition dispatch", func(t *testing.T) {
		topic := &changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig{
			DispatchType:      changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_ONE_TOPIC,
			ReplicationFactor: 1,
			PartitionNum:      1,
		}
		require.Error(t, validator.Validate(topic))
		topic.DefaultTopic = "events"
		require.NoError(t, validator.Validate(topic))

		dispatcher := &changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig_PartitionDispatcherConfig{
			PartitionType: changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig_COLUMNS,
			Matcher:       []string{"db.table"},
		}
		require.Error(t, validator.Validate(dispatcher))
		dispatcher.Columns = []string{"id"}
		require.NoError(t, validator.Validate(dispatcher))
	})

	t.Run("S3 endpoint", func(t *testing.T) {
		endpoint := &changefeedopenpb.S3DownstreamConfig_Endpoint{
			S3Uri:    "s3://bucket/prefix",
			AuthType: changefeedopenpb.S3DownstreamConfig_Endpoint_S3_AUTH_TYPE_AK_SK,
		}
		require.Error(t, validator.Validate(endpoint))
		endpoint.AccessKey = "access-key"
		require.NoError(t, validator.Validate(endpoint), "secret key is create-only")

		endpoint.AuthType = changefeedopenpb.S3DownstreamConfig_Endpoint_S3_AUTH_TYPE_ASSUME_ROLE
		endpoint.AccessKey = ""
		require.Error(t, validator.Validate(endpoint))
		endpoint.IamAssumedRoleArn = "arn:aws:iam::123456789012:role/cdc"
		require.NoError(t, validator.Validate(endpoint))

		endpoint.Provider = changefeedopenpb.S3DownstreamConfig_Endpoint_ALIBABA_OSS
		require.Error(t, validator.Validate(endpoint))
		endpoint.CustomEndpoint = "https://oss.example.com"
		require.NoError(t, validator.Validate(endpoint))
	})

	t.Run("Pulsar authentication and topic", func(t *testing.T) {
		auth := &changefeedopenpb.PulsarDownstreamConfig_Authentication{
			AuthType: changefeedopenpb.PulsarDownstreamConfig_Authentication_BASIC,
		}
		require.Error(t, validator.Validate(auth))
		auth.Username = "pulsar-user"
		require.NoError(t, validator.Validate(auth), "password is create-only")
		auth.AuthType = changefeedopenpb.PulsarDownstreamConfig_Authentication_TOKEN
		auth.Username = ""
		require.NoError(t, validator.Validate(auth), "token is create-only")

		topic := &changefeedopenpb.PulsarDownstreamConfig_TopicDistribution{
			Tenant:       "public",
			Namespace:    "default",
			DispatchType: changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_ONE_TOPIC,
		}
		require.Error(t, validator.Validate(topic))
		topic.DefaultTopic = "events"
		require.NoError(t, validator.Validate(topic))
	})
}

func TestCreateConditionalRequiredProtovalidate(t *testing.T) {
	validator, err := protovalidate.New()
	require.NoError(t, err)

	t.Run("MySQL password", func(t *testing.T) {
		changefeed := validBaseChangefeed(changefeedopenpb.DownstreamType_MYSQL)
		changefeed.DownstreamConfig = &changefeedopenpb.Changefeed_Mysql{Mysql: validMySQLDownstreamConfig()}
		req := &changefeedopenpb.CreateChangefeedRequest{Changefeed: changefeed}
		require.Error(t, validator.Validate(req))
		changefeed.GetMysql().GetConnection().Password = "secret"
		require.NoError(t, validator.Validate(req))
	})

	t.Run("Kafka and Schema Registry passwords", func(t *testing.T) {
		changefeed := validBaseChangefeed(changefeedopenpb.DownstreamType_KAFKA)
		kafka := validKafkaDownstreamConfig()
		kafka.Authentication.AuthType = changefeedopenpb.KafkaDownstreamConfig_Authentication_SASL_PLAIN
		kafka.Authentication.Username = stringPtr("kafka-user")
		changefeed.DownstreamConfig = &changefeedopenpb.Changefeed_Kafka{Kafka: kafka}
		req := &changefeedopenpb.CreateChangefeedRequest{Changefeed: changefeed}
		require.Error(t, validator.Validate(req))
		kafka.Authentication.Password = stringPtr("secret")
		require.NoError(t, validator.Validate(req))

		kafka.DataFormat = validKafkaAvroDataFormat()
		registry := kafka.DataFormat.GetAvroConfig().GetSchemaRegistry()
		registry.EnableBasicAuth = true
		registry.BasicAuthUsername = "schema-user"
		require.Error(t, validator.Validate(req))
		registry.BasicAuthPassword = "secret"
		require.NoError(t, validator.Validate(req))
	})

	t.Run("S3 URI and secret key", func(t *testing.T) {
		changefeed := validBaseChangefeed(changefeedopenpb.DownstreamType_S3)
		s3 := validS3DownstreamConfig()
		changefeed.DownstreamConfig = &changefeedopenpb.Changefeed_S3{S3: s3}
		req := &changefeedopenpb.CreateChangefeedRequest{Changefeed: changefeed}
		s3.Endpoint.S3Uri = ""
		require.Error(t, validator.Validate(req))
		s3.Endpoint.S3Uri = "s3://bucket/prefix"
		s3.Endpoint.SecretKey = ""
		require.Error(t, validator.Validate(req))
		s3.Endpoint.SecretKey = "secret"
		require.NoError(t, validator.Validate(req))
	})

	t.Run("Pulsar password and token", func(t *testing.T) {
		changefeed := validBaseChangefeed(changefeedopenpb.DownstreamType_PULSAR)
		pulsar := validPulsarDownstreamConfig()
		changefeed.DownstreamConfig = &changefeedopenpb.Changefeed_Pulsar{Pulsar: pulsar}
		req := &changefeedopenpb.CreateChangefeedRequest{Changefeed: changefeed}
		pulsar.Authentication.Password = ""
		require.Error(t, validator.Validate(req))
		pulsar.Authentication.Password = "secret"
		require.NoError(t, validator.Validate(req))
		pulsar.Authentication.AuthType = changefeedopenpb.PulsarDownstreamConfig_Authentication_TOKEN
		pulsar.Authentication.Username = ""
		pulsar.Authentication.Password = ""
		require.Error(t, validator.Validate(req))
		pulsar.Authentication.Token = "token"
		require.NoError(t, validator.Validate(req))
	})

	t.Run("downstream config and network-specific endpoints", func(t *testing.T) {
		changefeed := validBaseChangefeed(changefeedopenpb.DownstreamType_KAFKA)
		changefeed.DownstreamConfig = &changefeedopenpb.Changefeed_Mysql{Mysql: validMySQLDownstreamConfig()}
		require.Error(t, validator.Validate(changefeed))

		kafka := validKafkaDownstreamConfig()
		kafka.Broker.BrokerEndpoints = ""
		changefeed.DownstreamConfig = &changefeedopenpb.Changefeed_Kafka{Kafka: kafka}
		require.Error(t, validator.Validate(changefeed))
		changefeed.NetworkInfo = &changefeedopenpb.NetworkInfo{
			NetworkType:    changefeedopenpb.NetworkInfo_NETWORK_TYPE_PRIVATE_LINK,
			SinkEndpointId: uint64Ptr(1),
		}
		require.NoError(t, validator.Validate(changefeed))
	})
}

func TestConnectionConditionalRequiredProtovalidate(t *testing.T) {
	validator, err := protovalidate.New()
	require.NoError(t, err)

	mysql := validMySQLDownstreamConfig()
	req := &changefeedopenpb.TestConnectionRequest{
		TidbId:         "1",
		DownstreamType: changefeedopenpb.DownstreamType_MYSQL,
		DownstreamConfig: &changefeedopenpb.TestConnectionRequest_Mysql{
			Mysql: mysql,
		},
	}
	require.Error(t, validator.Validate(req), "password is required")
	mysql.Connection.Password = "secret"
	require.NoError(t, validator.Validate(req))

	mysql.Connection.Endpoint = ""
	require.Error(t, validator.Validate(req), "public MySQL endpoint is required")
	req.NetworkInfo = &changefeedopenpb.NetworkInfo{
		NetworkType:    changefeedopenpb.NetworkInfo_NETWORK_TYPE_PRIVATE_LINK,
		SinkEndpointId: uint64Ptr(1),
	}
	require.NoError(t, validator.Validate(req))

	req.DownstreamType = changefeedopenpb.DownstreamType_KAFKA
	require.Error(t, validator.Validate(req), "downstream config must match its type")
}

func TestEditRejectsMismatchedDownstreamConfig(t *testing.T) {
	adapter := &consoleChangefeedAdapter{}
	_, err := adapter.mergeChangefeedEdit(&dataflow.Changefeed{
		ChangefeedType: dataflow.ChangefeedType_KAFKA,
		Request:        &dataflow.ChangefeedCreateReq{KafkaInfo: &dataflow.KafkaInfo{}},
	}, &changefeedopenpb.EditChangefeedDownstreamConfigRequest{
		DownstreamType: changefeedopenpb.DownstreamType_KAFKA,
		DownstreamConfig: &changefeedopenpb.EditChangefeedDownstreamConfigRequest_Mysql{
			Mysql: &changefeedopenpb.MySQLDownstreamConfig{},
		},
	})
	require.Error(t, err)
}

func TestToOpenAPINetworkInfoPreservesPrivateLinkPorts(t *testing.T) {
	sinkEndpointID := uint64(42)
	got := toOpenAPINetworkInfo(&dataflow.NetworkInfo{
		NetworkType:    dataflow.NetworkType_NetworkType_Private_Link,
		SinkEndpointId: &sinkEndpointID,
		Ports:          []uint32{9092, 9093},
	})

	require.Equal(t, changefeedopenpb.NetworkInfo_NETWORK_TYPE_PRIVATE_LINK, got.GetNetworkType())
	require.Equal(t, sinkEndpointID, got.GetSinkEndpointId())
	require.Equal(t, []uint32{9092, 9093}, got.GetPorts())
}

func TestRefreshRealtimeCheckpoint(t *testing.T) {
	t.Run("preserves cached checkpoint before kernel creation", func(t *testing.T) {
		adapter := &consoleChangefeedAdapter{consoleChangefeedAPI: &consoleChangefeedStub{
			getChangefeedDetail: func(context.Context, *dataflow.ChangefeedID) (*structpb.Struct, error) {
				return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
			},
		}}
		changefeed := &dataflow.Changefeed{ChangefeedId: 1, CheckpointTso: 42}
		require.NoError(t, adapter.refreshRealtimeCheckpoint(context.Background(), changefeed))
		require.Equal(t, uint64(42), changefeed.GetCheckpointTso())
	})

	t.Run("preserves checkpoint milliseconds", func(t *testing.T) {
		checkpoint := time.Date(2026, time.July, 31, 3, 39, 2, 916*int(time.Millisecond), time.UTC)
		checkpointTSO := uint64(checkpoint.UnixMilli()) << 18
		detail, err := structpb.NewStruct(map[string]interface{}{
			"checkpoint_ts":   strconv.FormatUint(checkpointTSO, 10),
			"checkpoint_time": "2026-07-31 03:39:02.916",
		})
		require.NoError(t, err)
		adapter := &consoleChangefeedAdapter{consoleChangefeedAPI: &consoleChangefeedStub{
			getChangefeedDetail: func(context.Context, *dataflow.ChangefeedID) (*structpb.Struct, error) {
				return detail, nil
			},
		}}
		changefeed := &dataflow.Changefeed{ChangefeedId: 1}
		require.NoError(t, adapter.refreshRealtimeCheckpoint(context.Background(), changefeed))
		require.Equal(t, checkpointTSO, changefeed.GetCheckpointTso())
		require.Equal(t, checkpoint, changefeed.GetCheckpointUtc().AsTime())
	})
}

func TestCCUSpecificationConversion(t *testing.T) {
	tests := map[string]struct {
		internal   string
		openAPICAP string
	}{
		"base":     {internal: "4rcu", openAPICAP: "4ccu"},
		"smallest": {internal: "2rcu", openAPICAP: "2ccu"},
		"largest":  {internal: "384rcu", openAPICAP: "384ccu"},
		"empty":    {internal: "", openAPICAP: ""},
	}
	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			require.Equal(t, tc.openAPICAP, toOpenAPICCU(tc.internal))
			require.Equal(t, tc.internal, toConsoleSpecification(tc.openAPICAP))
		})
	}
}

func uint64Ptr(value uint64) *uint64 {
	return &value
}

func validBaseChangefeed(downstreamType changefeedopenpb.DownstreamType) *changefeedopenpb.Changefeed {
	return &changefeedopenpb.Changefeed{
		TidbId:             "1",
		Name:               "changefeed",
		ChangefeedCapacity: "4ccu",
		DownstreamType:     downstreamType,
		TableConfig:        &changefeedopenpb.TableConfig{Mode: changefeedopenpb.TableConfig_FORCE_SYNC},
		StartPosition:      &changefeedopenpb.StartPosition{Mode: changefeedopenpb.StartPosition_FROM_NOW},
		DownstreamConfig:   nil,
	}
}

func validMySQLDownstreamConfig() *changefeedopenpb.MySQLDownstreamConfig {
	return &changefeedopenpb.MySQLDownstreamConfig{
		Connection: &changefeedopenpb.Connection{
			Endpoint: "mysql.example.com:4000",
			Username: "cdc-user",
		},
	}
}

func validKafkaDownstreamConfig() *changefeedopenpb.KafkaDownstreamConfig {
	return &changefeedopenpb.KafkaDownstreamConfig{
		Broker: &changefeedopenpb.KafkaDownstreamConfig_Broker{
			Version:         changefeedopenpb.KafkaDownstreamConfig_KAFKA_VERSION_3XX,
			BrokerEndpoints: "broker.example.com:9092",
		},
		Authentication: &changefeedopenpb.KafkaDownstreamConfig_Authentication{
			AuthType: changefeedopenpb.KafkaDownstreamConfig_Authentication_DISABLE,
		},
		DataFormat: &changefeedopenpb.KafkaDownstreamConfig_DataFormat{
			Protocol: changefeedopenpb.KafkaDownstreamConfig_DataFormat_PROTOCOL_CANAL_JSON,
		},
		TopicPartitionConfig: &changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig{
			DispatchType:      changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_BY_TABLE,
			ReplicationFactor: 1,
			PartitionNum:      1,
		},
	}
}

func validKafkaAvroDataFormat() *changefeedopenpb.KafkaDownstreamConfig_DataFormat {
	return &changefeedopenpb.KafkaDownstreamConfig_DataFormat{
		Protocol: changefeedopenpb.KafkaDownstreamConfig_DataFormat_PROTOCOL_AVRO,
		Config: &changefeedopenpb.KafkaDownstreamConfig_DataFormat_AvroConfig_{
			AvroConfig: &changefeedopenpb.KafkaDownstreamConfig_DataFormat_AvroConfig{
				SchemaRegistry: &changefeedopenpb.KafkaDownstreamConfig_DataFormat_AvroConfig_SchemaRegistry{
					SchemaRegistryEndpoints: "https://registry.example.com",
				},
			},
		},
	}
}

func validS3DownstreamConfig() *changefeedopenpb.S3DownstreamConfig {
	return &changefeedopenpb.S3DownstreamConfig{
		Endpoint: &changefeedopenpb.S3DownstreamConfig_Endpoint{
			S3Uri:     "s3://bucket/prefix",
			AuthType:  changefeedopenpb.S3DownstreamConfig_Endpoint_S3_AUTH_TYPE_AK_SK,
			AccessKey: "access-key",
			SecretKey: "secret",
			Provider:  changefeedopenpb.S3DownstreamConfig_Endpoint_AWS,
		},
		DataFormat: &changefeedopenpb.S3DataFormat{
			Protocol: changefeedopenpb.S3DataFormat_PROTOCOL_CANAL_JSON,
		},
	}
}

func validPulsarDownstreamConfig() *changefeedopenpb.PulsarDownstreamConfig {
	return &changefeedopenpb.PulsarDownstreamConfig{
		DestinationProtocol: changefeedopenpb.PulsarDownstreamConfig_PULSAR,
		Endpoint:            "pulsar.example.com:6650",
		Authentication: &changefeedopenpb.PulsarDownstreamConfig_Authentication{
			AuthType: changefeedopenpb.PulsarDownstreamConfig_Authentication_BASIC,
			Username: "pulsar-user",
			Password: "secret",
		},
		DataFormat: &changefeedopenpb.PulsarDownstreamConfig_DataFormat{
			Protocol: changefeedopenpb.PulsarDownstreamConfig_DataFormat_CANAL_JSON,
		},
		TopicDistribution: &changefeedopenpb.PulsarDownstreamConfig_TopicDistribution{
			Tenant:       "public",
			Namespace:    "default",
			DispatchType: changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_BY_TABLE,
		},
		PartitionDispatchers: []*changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig_PartitionDispatcherConfig{
			{
				PartitionType: changefeedopenpb.KafkaDownstreamConfig_TopicPartitionConfig_TABLE,
				Matcher:       []string{"*.*"},
			},
		},
		AdvancedSetting: &changefeedopenpb.PulsarDownstreamConfig_AdvancedSetting{
			BatchingMaxMessages:     1,
			BatchingMaxPublishDelay: 1,
			ConnectionTimeout:       1,
			OperationTimeout:        1,
			SendTimeout:             1,
		},
	}
}