package cvt

import (
	"encoding/json"
	"fmt"
	"strconv"
	"strings"

	dataflowng "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	ccapi "github.com/tidbcloud/devops/api/cloudconfig"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
)

func (c fromDataflow) FullChangefeed(detail *dataflow.Connector) (*ccapi.InternalChangefeed, error) {
	out := c.Changefeed(detail)
	if err := populateChangefeedMeta(out, detail); err != nil {
		return nil, err
	}
	return out, nil
}

func (c fromDataflowNg) FullChangefeed(detail *dataflowng.Changefeed) (*ccapi.InternalChangefeed, error) {
	out := c.Changefeed(detail)
	if err := populateNgChangefeedMeta(out, detail); err != nil {
		return nil, err
	}
	return out, nil
}

func populateChangefeedMeta(out *ccapi.InternalChangefeed, detail *dataflow.Connector) error {
	if out == nil || detail == nil || detail.GetRequest() == nil {
		return fmt.Errorf("changefeed detail is incomplete")
	}

	req := detail.GetRequest()
	out.NetworkInfo = buildChangefeedNetworkInfo(req.GetNetworkInfo())
	out.CheckpointTSO = formatTSO(detail.GetCheckpointTso())
	if detail.GetCheckpointUtc() != nil {
		out.CheckpointTS = detail.GetCheckpointUtc().AsTime().Unix()
	}
	out.RCU = detail.GetChangefeedInstanceType()
	if out.RCU == "" {
		out.RCU = req.GetChangefeedInstanceType()
	}
	out.TableConfig = buildChangefeedTableConfig(changefeedTableConfig(req, detail.GetConnectorType()))
	out.StartPosition = buildChangefeedStartPosition(changefeedStartPosition(req, detail.GetConnectorType()))

	config, err := SanitizeDownstreamConfig(changefeedDownstreamConfig(req, detail.GetConnectorType()))
	if err != nil {
		return fmt.Errorf("sanitize downstream config: %w", err)
	}
	out.DownstreamConfig = config
	return nil
}

func populateNgChangefeedMeta(out *ccapi.InternalChangefeed, detail *dataflowng.Changefeed) error {
	if out == nil || detail == nil || detail.GetRequest() == nil {
		return fmt.Errorf("changefeed detail is incomplete")
	}

	req := detail.GetRequest()
	out.NetworkInfo = buildNgChangefeedNetworkInfo(req.GetNetworkInfo())
	out.CheckpointTSO = formatTSO(detail.GetCheckpointTso())
	if detail.GetCheckpointUtc() != nil {
		out.CheckpointTS = detail.GetCheckpointUtc().AsTime().Unix()
	}
	out.RCU = detail.GetChangefeedSpec()
	if out.RCU == "" {
		out.RCU = req.GetChangefeedSpec()
	}
	out.TableConfig = buildNgChangefeedTableConfig(ngChangefeedTableConfig(req, detail.GetChangefeedType()))
	out.StartPosition = buildNgChangefeedStartPosition(ngChangefeedStartPosition(req, detail.GetChangefeedType()))

	config, err := SanitizeDownstreamConfig(ngChangefeedDownstreamConfig(req, detail.GetChangefeedType()))
	if err != nil {
		return fmt.Errorf("sanitize downstream config: %w", err)
	}
	out.DownstreamConfig = config
	return nil
}

func formatTSO(tso uint64) string {
	if tso == 0 {
		return ""
	}
	return strconv.FormatUint(tso, 10)
}

func buildChangefeedNetworkInfo(in *dataflow.NetworkInfo) *ccapi.ChangefeedNetworkInfo {
	if in == nil {
		return nil
	}
	out := &ccapi.ChangefeedNetworkInfo{
		NetworkType:            normalizeEnum(in.GetNetworkType().String(), "NetworkType_"),
		PrivateLinkServiceName: in.GetPrivateLinkServiceName(),
		KafkaID:                in.GetKafkaId(),
		Ports:                  append([]uint32(nil), in.GetPorts()...),
		AdvertisedPatternGroup: cloneStringMap(in.GetAdvertisedPatternGroup()),
	}
	if in.SinkEndpointId != nil {
		out.SinkEndpointID = strconv.FormatUint(in.GetSinkEndpointId(), 10)
	}
	return out
}

func buildNgChangefeedNetworkInfo(in *dataflowng.NetworkInfo) *ccapi.ChangefeedNetworkInfo {
	if in == nil {
		return nil
	}
	out := &ccapi.ChangefeedNetworkInfo{
		NetworkType: normalizeEnum(in.GetNetworkType().String(), "NetworkType_"),
		Ports:       append([]uint32(nil), in.GetPorts()...),
	}
	if in.SinkEndpointId != nil {
		out.SinkEndpointID = strconv.FormatUint(in.GetSinkEndpointId(), 10)
	}
	return out
}

func cloneStringMap(in map[string]string) map[string]string {
	if len(in) == 0 {
		return nil
	}
	out := make(map[string]string, len(in))
	for key, value := range in {
		out[key] = value
	}
	return out
}

func buildChangefeedTableConfig(in *dataflow.TableConfig) *ccapi.ChangefeedTableConfig {
	if in == nil {
		return nil
	}
	out := &ccapi.ChangefeedTableConfig{
		FilterRules:   append([]string{}, in.GetFilterRule()...),
		Mode:          normalizeEnum(in.GetMode().String(), "TableMode_"),
		CaseSensitive: in.GetCaseSensitive(),
		EventFilters:  make([]*ccapi.ChangefeedEventFilter, 0, len(in.GetEventFilterRule())),
	}
	for _, rule := range in.GetEventFilterRule() {
		if rule == nil {
			continue
		}
		out.EventFilters = append(out.EventFilters, &ccapi.ChangefeedEventFilter{
			TableMatchers:                   append([]string{}, rule.GetMatcher()...),
			IgnoredEvents:                   append([]string{}, rule.GetIgnoreEvent()...),
			IgnoredSQLStatements:            append([]string{}, rule.GetIgnoreSql()...),
			IgnoredInsertValueExpression:    rule.GetIgnoreInsertValueExpr(),
			IgnoredUpdateNewValueExpression: rule.GetIgnoreUpdateNewValueExpr(),
			IgnoredUpdateOldValueExpression: rule.GetIgnoreUpdateOldValueExpr(),
			IgnoredDeleteValueExpression:    rule.GetIgnoreDeleteValueExpr(),
		})
	}
	return out
}

func buildNgChangefeedTableConfig(in *dataflowng.TableConfig) *ccapi.ChangefeedTableConfig {
	if in == nil {
		return nil
	}
	out := &ccapi.ChangefeedTableConfig{
		FilterRules:   append([]string{}, in.GetFilterRule()...),
		Mode:          normalizeEnum(in.GetMode().String(), "TableMode_"),
		CaseSensitive: in.GetCaseSensitive(),
		EventFilters:  make([]*ccapi.ChangefeedEventFilter, 0, len(in.GetEventFilterRule())),
	}
	for _, rule := range in.GetEventFilterRule() {
		if rule == nil {
			continue
		}
		out.EventFilters = append(out.EventFilters, &ccapi.ChangefeedEventFilter{
			TableMatchers:                   append([]string{}, rule.GetMatcher()...),
			IgnoredEvents:                   append([]string{}, rule.GetIgnoreEvent()...),
			IgnoredSQLStatements:            append([]string{}, rule.GetIgnoreSql()...),
			IgnoredInsertValueExpression:    rule.GetIgnoreInsertValueExpr(),
			IgnoredUpdateNewValueExpression: rule.GetIgnoreUpdateNewValueExpr(),
			IgnoredUpdateOldValueExpression: rule.GetIgnoreUpdateOldValueExpr(),
			IgnoredDeleteValueExpression:    rule.GetIgnoreDeleteValueExpr(),
		})
	}
	return out
}

func buildChangefeedStartPosition(in *dataflow.StartPosition) *ccapi.ChangefeedStartPosition {
	if in == nil {
		return nil
	}
	return &ccapi.ChangefeedStartPosition{
		Mode:      normalizeEnum(in.GetMode().String(), "StartMode_"),
		TSO:       formatTSO(in.GetTso()),
		Timestamp: in.GetUtc(),
	}
}

func buildNgChangefeedStartPosition(in *dataflowng.StartPosition) *ccapi.ChangefeedStartPosition {
	if in == nil {
		return nil
	}
	return &ccapi.ChangefeedStartPosition{
		Mode:      normalizeEnum(in.GetMode().String(), "StartMode_"),
		TSO:       formatTSO(in.GetTso()),
		Timestamp: in.GetUtc(),
	}
}

func normalizeEnum(value, prefix string) string {
	return strings.ToUpper(strings.TrimPrefix(value, prefix))
}

func changefeedTableConfig(req *dataflow.ConnectorCreateReq, typ dataflow.ConnectorType) *dataflow.TableConfig {
	switch typ {
	case dataflow.ConnectorType_KAFKA:
		return req.GetKafkaInfo().GetTableConfig()
	case dataflow.ConnectorType_TIDB, dataflow.ConnectorType_MYSQL:
		return req.GetMysqlInfo().GetTableConfig()
	case dataflow.ConnectorType_SERVERLESS_TIDB:
		return req.GetServerlessTidbInfo().GetMysqlInfo().GetTableConfig()
	case dataflow.ConnectorType_S3:
		return req.GetS3Info().GetTableConfig()
	case dataflow.ConnectorType_GCS:
		return req.GetGcsInfo().GetTableConfig()
	case dataflow.ConnectorType_PULSAR:
		return req.GetPulsarInfo().GetTableConfig()
	case dataflow.ConnectorType_AZURE_BLOB_STORAGE:
		return req.GetAzureBlobStorageInfo().GetTableConfig()
	default:
		return nil
	}
}

func changefeedStartPosition(req *dataflow.ConnectorCreateReq, typ dataflow.ConnectorType) *dataflow.StartPosition {
	switch typ {
	case dataflow.ConnectorType_KAFKA:
		return req.GetKafkaInfo().GetStartPosition()
	case dataflow.ConnectorType_TIDB, dataflow.ConnectorType_MYSQL:
		return req.GetMysqlInfo().GetStartPosition()
	case dataflow.ConnectorType_SERVERLESS_TIDB:
		return req.GetServerlessTidbInfo().GetMysqlInfo().GetStartPosition()
	case dataflow.ConnectorType_S3:
		return req.GetS3Info().GetStartPosition()
	case dataflow.ConnectorType_GCS:
		return req.GetGcsInfo().GetStartPosition()
	case dataflow.ConnectorType_PULSAR:
		return req.GetPulsarInfo().GetStartPosition()
	case dataflow.ConnectorType_AZURE_BLOB_STORAGE:
		return req.GetAzureBlobStorageInfo().GetStartPosition()
	default:
		return nil
	}
}

func changefeedDownstreamConfig(req *dataflow.ConnectorCreateReq, typ dataflow.ConnectorType) proto.Message {
	switch typ {
	case dataflow.ConnectorType_KAFKA:
		return req.GetKafkaInfo()
	case dataflow.ConnectorType_TIDB, dataflow.ConnectorType_MYSQL:
		return req.GetMysqlInfo()
	case dataflow.ConnectorType_SERVERLESS_TIDB:
		return req.GetServerlessTidbInfo()
	case dataflow.ConnectorType_S3:
		return req.GetS3Info()
	case dataflow.ConnectorType_GCS:
		return req.GetGcsInfo()
	case dataflow.ConnectorType_PULSAR:
		return req.GetPulsarInfo()
	case dataflow.ConnectorType_AZURE_BLOB_STORAGE:
		return req.GetAzureBlobStorageInfo()
	default:
		return nil
	}
}

func ngChangefeedTableConfig(req *dataflowng.ChangefeedCreateReq, typ dataflowng.ChangefeedType) *dataflowng.TableConfig {
	switch typ {
	case dataflowng.ChangefeedType_KAFKA:
		return req.GetKafkaInfo().GetTableConfig()
	case dataflowng.ChangefeedType_TIDB, dataflowng.ChangefeedType_MYSQL:
		return req.GetMysqlInfo().GetTableConfig()
	case dataflowng.ChangefeedType_SERVERLESS_TIDB:
		return req.GetServerlessTidbInfo().GetMysqlInfo().GetTableConfig()
	case dataflowng.ChangefeedType_S3:
		return req.GetS3Info().GetTableConfig()
	case dataflowng.ChangefeedType_GCS:
		return req.GetGcsInfo().GetTableConfig()
	case dataflowng.ChangefeedType_PULSAR:
		return req.GetPulsarInfo().GetTableConfig()
	default:
		return nil
	}
}

func ngChangefeedStartPosition(req *dataflowng.ChangefeedCreateReq, typ dataflowng.ChangefeedType) *dataflowng.StartPosition {
	switch typ {
	case dataflowng.ChangefeedType_KAFKA:
		return req.GetKafkaInfo().GetStartPosition()
	case dataflowng.ChangefeedType_TIDB, dataflowng.ChangefeedType_MYSQL:
		return req.GetMysqlInfo().GetStartPosition()
	case dataflowng.ChangefeedType_SERVERLESS_TIDB:
		return req.GetServerlessTidbInfo().GetMysqlInfo().GetStartPosition()
	case dataflowng.ChangefeedType_S3:
		return req.GetS3Info().GetStartPosition()
	case dataflowng.ChangefeedType_GCS:
		return req.GetGcsInfo().GetStartPosition()
	case dataflowng.ChangefeedType_PULSAR:
		return req.GetPulsarInfo().GetStartPosition()
	default:
		return nil
	}
}

func ngChangefeedDownstreamConfig(req *dataflowng.ChangefeedCreateReq, typ dataflowng.ChangefeedType) proto.Message {
	switch typ {
	case dataflowng.ChangefeedType_KAFKA:
		return req.GetKafkaInfo()
	case dataflowng.ChangefeedType_TIDB, dataflowng.ChangefeedType_MYSQL:
		return req.GetMysqlInfo()
	case dataflowng.ChangefeedType_SERVERLESS_TIDB:
		return req.GetServerlessTidbInfo()
	case dataflowng.ChangefeedType_S3:
		return req.GetS3Info()
	case dataflowng.ChangefeedType_GCS:
		return req.GetGcsInfo()
	case dataflowng.ChangefeedType_PULSAR:
		return req.GetPulsarInfo()
	default:
		return nil
	}
}

// SanitizeDownstreamConfig converts a protobuf sink configuration to its API
// representation and removes credentials and duplicated table/start settings.
func SanitizeDownstreamConfig(message proto.Message) (map[string]interface{}, error) {
	if message == nil || !message.ProtoReflect().IsValid() {
		return nil, nil
	}
	// Configuration defaults are effective values, not missing data. This is
	// especially important for zero-valued enums such as CANAL_JSON, TABLE and
	// PRECISE, as well as false-valued TLS/extension switches.
	data, err := (protojson.MarshalOptions{
		UseProtoNames:     true,
		EmitDefaultValues: true,
	}).Marshal(message)
	if err != nil {
		return nil, err
	}
	var config map[string]interface{}
	if err := json.Unmarshal(data, &config); err != nil {
		return nil, err
	}
	sanitizeConfigMap(config)
	if len(config) == 0 {
		return nil, nil
	}
	return config, nil
}

func sanitizeConfigMap(config map[string]interface{}) {
	for key, value := range config {
		switch {
		case key == "table_config", key == "start_position", isSensitiveConfigKey(key):
			delete(config, key)
		default:
			sanitizeConfigValue(value)
		}
	}
}

func sanitizeConfigValue(value interface{}) {
	switch typed := value.(type) {
	case map[string]interface{}:
		sanitizeConfigMap(typed)
	case []interface{}:
		for _, item := range typed {
			sanitizeConfigValue(item)
		}
	}
}

func isSensitiveConfigKey(key string) bool {
	normalized := strings.Map(func(r rune) rune {
		if r >= 'A' && r <= 'Z' {
			return r + ('a' - 'A')
		}
		if r >= 'a' && r <= 'z' {
			return r
		}
		return -1
	}, key)
	for _, suffix := range []string{
		"password",
		"token",
		"accesskey",
		"secretkey",
		"privatekey",
		"clientsecret",
		"credential",
		"credentials",
	} {
		if strings.HasSuffix(normalized, suffix) {
			return true
		}
	}
	return false
}