package cloudconfig

import (
	"encoding/json"
	"strings"
	"time"

	dataflowng "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/devops/pkg/service/cloudconfig/cvt"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/timestamppb"

	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
)

var _ = Describe("[ChangefeedMeta]", func() {
	DescribeTable("validates operation cluster types",
		func(clusterType string, shouldSucceed bool) {
			err := validateChangefeedOperationClusterType(clusterType)
			if shouldSucceed {
				Expect(err).NotTo(HaveOccurred())
				return
			}
			Expect(err).To(HaveOccurred())
		},
		Entry("dedicated changefeed", "CHANGEFEED", true),
		Entry("next-gen changefeed", "CHANGEFEED_NG", true),
		Entry("essential changefeed", "CHANGEFEED_ESSENTIAL", true),
		Entry("unsupported cluster", "DEDICATED_TIDB", false),
	)

	It("keeps kernel IDs empty until internal changefeeds are matched", func() {
		out := cvt.From.Dataflow.Changefeed(&dataflow.Connector{
			ConnectorId:   201,
			ConnectorType: dataflow.ConnectorType_KAFKA,
			Status:        dataflow.Connector_RUNNING,
			Name:          "dedicated-changefeed",
		})
		Expect(out.Id).To(BeEmpty())
		Expect(out.DisplayId).To(Equal("201"))
		Expect(out.ClusterType).To(Equal("CHANGEFEED"))
		Expect(out.Status).To(Equal("normal"))
		Expect(out.DownstreamType).To(Equal("kafka"))
		Expect(out.TicdcCluster).To(BeNil())

		ngOut := cvt.From.DataflowNg.Changefeed(&dataflowng.Changefeed{
			ChangefeedId:   202,
			ChangefeedType: dataflowng.ChangefeedType_MYSQL,
			Status:         dataflowng.Changefeed_CREATING,
			Name:           "premium-changefeed",
		})
		Expect(ngOut.Id).To(BeEmpty())
		Expect(ngOut.DisplayId).To(Equal("202"))
		Expect(ngOut.ClusterType).To(Equal("CHANGEFEED_NG"))
		Expect(ngOut.Status).To(Equal("creating"))
		Expect(ngOut.DownstreamType).To(Equal("mysql"))
		Expect(ngOut.TicdcCluster).To(BeNil())
	})

	It("matches internal changefeeds by business display ID for TiCDC enrichment", func() {
		dedicatedInternal := &dataflow.InternalChangefeed{
			Id: 301, DisplayId: 201, TicdcCluster: &dataflow.InternalTiCDCCluster{Id: 401, Name: "ticdc"},
		}
		dedicatedAPI := cvt.From.Dataflow.Changefeed(&dataflow.Connector{ConnectorId: 201})
		cvt.From.Dataflow.EnrichChangefeed(dedicatedAPI, dedicatedInternal)
		Expect(dedicatedAPI.Id).To(Equal("301"))
		Expect(dedicatedAPI.DisplayId).To(Equal("201"))
		Expect(dedicatedAPI.TicdcCluster.Id).To(Equal("401"))

		nextGenInternal := &dataflowng.InternalChangefeed{
			Id: 302, DisplayId: 202, TicdcCluster: &dataflowng.InternalTiCDCCluster{Id: 402, Name: "ticdc-ng"},
		}
		nextGenAPI := cvt.From.DataflowNg.Changefeed(&dataflowng.Changefeed{ChangefeedId: 202})
		cvt.From.DataflowNg.EnrichChangefeed(nextGenAPI, nextGenInternal)
		Expect(nextGenAPI.Id).To(Equal("302"))
		Expect(nextGenAPI.DisplayId).To(Equal("202"))
		Expect(nextGenAPI.TicdcCluster.Id).To(Equal("402"))
	})

	It("normalizes dedicated changefeed metadata and redacts credentials", func() {
		sinkEndpointID := uint64(9007199254740993)
		checkpointTime := time.Unix(1_700_000_000, 0)
		detail := &dataflow.Connector{
			ConnectorId:   101,
			ConnectorType: dataflow.ConnectorType_KAFKA,
			CheckpointTso: 452348901234567890,
			CheckpointUtc: timestamppb.New(checkpointTime),
			Request: &dataflow.ConnectorCreateReq{
				ClusterId:              201,
				ChangefeedInstanceType: "4rcu",
				NetworkInfo: &dataflow.NetworkInfo{
					NetworkType:            dataflow.NetworkType_NetworkType_Private_Link,
					PrivateLinkServiceName: "com.amazonaws.vpce.svc",
					KafkaId:                "kafka-1",
					SinkEndpointId:         &sinkEndpointID,
					Ports:                  []uint32{9092},
					AdvertisedPatternGroup: map[string]string{"broker": "advertised.example.com"},
				},
				KafkaInfo: &dataflow.KafkaInfo{
					Broker: &dataflow.Broker{BrokerEndpoints: "broker.example.com:9092"},
					Authentication: &dataflow.Authentication{
						UserName: "cdc-user",
						Password: "kafka-password",
					},
					DataFormat: &dataflow.DataFormat{
						Protocol: dataflow.Protocol_CANAL_JSON,
						SchemaRegistry: &dataflow.SchemaRegistry{
							SchemaRegistryEndpoints: "registry.example.com",
							UserName:                "registry-user",
							Password:                "registry-password",
						},
					},
					TableConfig: &dataflow.TableConfig{
						FilterRule:    []string{"db.*"},
						Mode:          dataflow.TableMode_FORCE_SYNC,
						CaseSensitive: true,
						EventFilterRule: []*dataflow.EventFilterRule{{
							Matcher:     []string{"db.orders"},
							IgnoreEvent: []string{"delete"},
						}},
					},
					StartPosition: &dataflow.StartPosition{
						Mode: dataflow.StartMode_FROM_TSO,
						Tso:  452348901234567000,
					},
				},
			},
		}

		out, err := cvt.From.Dataflow.FullChangefeed(detail)
		Expect(err).NotTo(HaveOccurred())
		Expect(out.CheckpointTSO).To(Equal("452348901234567890"))
		Expect(out.CheckpointTS).To(Equal(checkpointTime.Unix()))
		Expect(out.RCU).To(Equal("4rcu"))
		Expect(out.NetworkInfo.NetworkType).To(Equal("PRIVATE_LINK"))
		Expect(out.NetworkInfo.PrivateLinkServiceName).To(Equal("com.amazonaws.vpce.svc"))
		Expect(out.NetworkInfo.KafkaID).To(Equal("kafka-1"))
		Expect(out.NetworkInfo.AdvertisedPatternGroup).To(Equal(map[string]string{"broker": "advertised.example.com"}))
		Expect(out.NetworkInfo.SinkEndpointID).To(Equal("9007199254740993"))
		Expect(out.TableConfig.FilterRules).To(Equal([]string{"db.*"}))
		Expect(out.TableConfig.Mode).To(Equal("FORCE_SYNC"))
		Expect(out.StartPosition.TSO).To(Equal("452348901234567000"))

		encoded, err := json.Marshal(out.DownstreamConfig)
		Expect(err).NotTo(HaveOccurred())
		config := string(encoded)
		Expect(config).To(ContainSubstring("broker.example.com:9092"))
		Expect(config).To(ContainSubstring("cdc-user"))
		Expect(config).To(ContainSubstring(`"protocol":"CANAL_JSON"`))
		Expect(config).To(ContainSubstring(`"decimal_handling_mode":"DECIMAL_HANDLING_MODE_PRECISE"`))
		Expect(config).To(ContainSubstring(`"bigint_unsigned_handling_mode":"BIGINT_UNSIGNED_HANDLING_MODE_LONG"`))
		Expect(config).To(ContainSubstring(`"tls_enable":false`))
		Expect(config).NotTo(ContainSubstring("kafka-password"))
		Expect(config).NotTo(ContainSubstring("registry-password"))
		Expect(config).NotTo(ContainSubstring("table_config"))
		Expect(config).NotTo(ContainSubstring("start_position"))
	})

	It("normalizes next-gen changefeed metadata", func() {
		detail := &dataflowng.Changefeed{
			ChangefeedId:   102,
			ChangefeedType: dataflowng.ChangefeedType_MYSQL,
			CheckpointTso:  452348901234567891,
			ChangefeedSpec: "8rcu",
			Request: &dataflowng.ChangefeedCreateReq{
				ClusterId: 202,
				MysqlInfo: &dataflowng.MySQLInfo{
					Connection: &dataflowng.Connection{
						Endpoint: "mysql.example.com:3306",
						Username: "cdc-user",
						Password: "mysql-password",
					},
					TableConfig:   &dataflowng.TableConfig{FilterRule: []string{"db.orders"}},
					StartPosition: &dataflowng.StartPosition{Mode: dataflowng.StartMode_FROM_UTC, Utc: "2026-07-01T00:00:00Z"},
				},
			},
		}

		out, err := cvt.From.DataflowNg.FullChangefeed(detail)
		Expect(err).NotTo(HaveOccurred())
		Expect(out.CheckpointTSO).To(Equal("452348901234567891"))
		Expect(out.RCU).To(Equal("8rcu"))
		Expect(out.StartPosition.Timestamp).To(Equal("2026-07-01T00:00:00Z"))

		encoded, err := json.Marshal(out.DownstreamConfig)
		Expect(err).NotTo(HaveOccurred())
		config := string(encoded)
		Expect(config).To(ContainSubstring("mysql.example.com:3306"))
		Expect(config).To(ContainSubstring("cdc-user"))
		Expect(config).NotTo(ContainSubstring("mysql-password"))
	})

	It("preserves zero-valued protocols for next-gen downstream config", func() {
		config, err := cvt.SanitizeDownstreamConfig(&dataflowng.KafkaInfo{
			DataFormat: &dataflowng.DataFormat{
				Protocol:       dataflowng.Protocol_CANAL_JSON,
				DebeziumConfig: &dataflowng.DebeziumConfig{OutputOldValue: true},
			},
		})
		Expect(err).NotTo(HaveOccurred())
		Expect(config).To(HaveKeyWithValue("data_format", And(
			HaveKeyWithValue("protocol", "CANAL_JSON"),
			HaveKeyWithValue("enable_tidb_extension", false),
			HaveKeyWithValue("decimal_handling_mode", "DECIMAL_HANDLING_MODE_PRECISE"),
			HaveKeyWithValue("bigint_unsigned_handling_mode", "BIGINT_UNSIGNED_HANDLING_MODE_LONG"),
			HaveKey("debezium_config"),
		)))
	})

	DescribeTable("redacts sink-specific credentials",
		func(message proto.Message, secrets ...string) {
			config, err := cvt.SanitizeDownstreamConfig(message)
			Expect(err).NotTo(HaveOccurred())
			encoded, err := json.Marshal(config)
			Expect(err).NotTo(HaveOccurred())
			result := string(encoded)
			for _, secret := range secrets {
				Expect(result).NotTo(ContainSubstring(secret))
			}
			Expect(strings.ToLower(result)).NotTo(ContainSubstring(`"password"`))
			Expect(strings.ToLower(result)).NotTo(ContainSubstring(`"token"`))
			Expect(strings.ToLower(result)).NotTo(ContainSubstring(`"access_key"`))
			Expect(strings.ToLower(result)).NotTo(ContainSubstring(`"secret_key"`))
		},
		Entry("S3", &dataflow.S3Info{
			Endpoint: &dataflow.S3Endpoint{
				S3Uri:     "s3://bucket/path",
				AccessKey: "access-key-value",
				SecretKey: "secret-key-value",
			},
		}, "access-key-value", "secret-key-value"),
		Entry("Azure Blob", &dataflow.AzureBlobStorageInfo{
			Endpoint: &dataflow.AzureBlobStorageEndpoint{
				BlobUri:  "https://account.blob.core.windows.net/container",
				SasToken: "sas-token-value",
			},
		}, "sas-token-value"),
		Entry("Pulsar", &dataflow.PulsarInfo{
			Authentication: &dataflow.PulsarAuthentication{
				UserName: "pulsar-user",
				Password: "pulsar-password",
				Token:    "pulsar-token",
			},
		}, "pulsar-password", "pulsar-token"),
	)
})