package ticdc

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

	dataflowng "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	serverlessrpc "github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
)

func dedicatedChangefeed(item *dataflow.Connector) *Changefeed {
	if item == nil {
		return nil
	}
	createdAt := int64(0)
	if item.CreatedAt != nil {
		createdAt = item.CreatedAt.AsTime().Unix()
	}
	return &Changefeed{
		DisplayID:      strconv.FormatUint(item.GetConnectorId(), 10),
		ClusterType:    "CHANGEFEED",
		Arch:           string(ArchitectureDedicated),
		DisplayName:    item.GetName(),
		DownstreamType: strings.ToLower(item.GetConnectorType().String()),
		Status:         dedicatedStatus(item.GetStatus()),
		CreatedAt:      createdAt,
	}
}

func nextGenChangefeed(item *dataflowng.Changefeed) *Changefeed {
	if item == nil {
		return nil
	}
	createdAt := int64(0)
	if item.CreatedAt != nil {
		createdAt = item.CreatedAt.AsTime().Unix()
	}
	return &Changefeed{
		DisplayID:      strconv.FormatUint(item.GetChangefeedId(), 10),
		ClusterType:    "CHANGEFEED_NG",
		Arch:           string(ArchitectureNextGen),
		DisplayName:    item.GetName(),
		DownstreamType: strings.ToLower(item.GetChangefeedType().String()),
		Status:         nextGenStatus(item.GetStatus()),
		CreatedAt:      createdAt,
	}
}

func essentialChangefeed(item *serverlessrpc.ChangefeedInfo) *Changefeed {
	if item == nil {
		return nil
	}
	result := &Changefeed{
		ID:               item.GetChangefeedId(),
		DisplayID:        item.GetChangefeedId(),
		ClusterType:      "CHANGEFEED_ESSENTIAL",
		Arch:             item.GetArch(),
		DisplayName:      item.GetDisplayName(),
		DownstreamType:   strings.ToLower(item.GetSinkType()),
		Status:           strings.ToLower(item.GetState()),
		CreatedAt:        parseEssentialTimestamp(item.GetCreatedAt()),
		TiCDCClusterName: item.GetTicdcClusterName(),
		CheckpointTSO:    formatTSO(item.GetCheckpointTso()),
		CheckpointTS:     parseEssentialTimestamp(item.GetCheckpointTime()),
		RCU:              strconv.FormatUint(uint64(item.GetRcu()), 10),
	}
	if item.GetStartTso() != 0 {
		result.StartPosition = &StartPosition{Mode: "tso", TSO: formatTSO(item.GetStartTso())}
	}
	return result
}

func parseEssentialTimestamp(value string) int64 {
	value = strings.TrimSpace(value)
	if value == "" {
		return 0
	}
	for _, layout := range []string{time.RFC3339Nano, "2006-01-02 15:04:05"} {
		if parsed, err := time.Parse(layout, value); err == nil {
			return parsed.Unix()
		}
	}
	return 0
}

func fullDedicatedChangefeed(item *dataflow.Connector) (*Changefeed, error) {
	result := dedicatedChangefeed(item)
	if result == nil || item.GetRequest() == nil {
		return nil, fmt.Errorf("changefeed detail is incomplete")
	}
	request := item.GetRequest()
	result.NetworkInfo = dedicatedNetworkInfo(request.GetNetworkInfo())
	result.CheckpointTSO = formatTSO(item.GetCheckpointTso())
	if item.GetCheckpointUtc() != nil {
		result.CheckpointTS = item.GetCheckpointUtc().AsTime().Unix()
	}
	result.RCU = item.GetChangefeedInstanceType()
	if result.RCU == "" {
		result.RCU = request.GetChangefeedInstanceType()
	}
	result.TableConfig = dedicatedTableConfig(dedicatedTableConfigSource(request, item.GetConnectorType()))
	result.StartPosition = dedicatedStartPosition(dedicatedStartPositionSource(request, item.GetConnectorType()))
	config, err := sanitizeDownstreamConfig(dedicatedDownstreamSource(request, item.GetConnectorType()))
	if err != nil {
		return nil, err
	}
	result.DownstreamConfig = config
	return result, nil
}

func fullNextGenChangefeed(item *dataflowng.Changefeed) (*Changefeed, error) {
	result := nextGenChangefeed(item)
	if result == nil || item.GetRequest() == nil {
		return nil, fmt.Errorf("changefeed detail is incomplete")
	}
	request := item.GetRequest()
	result.NetworkInfo = nextGenNetworkInfo(request.GetNetworkInfo())
	result.CheckpointTSO = formatTSO(item.GetCheckpointTso())
	if item.GetCheckpointUtc() != nil {
		result.CheckpointTS = item.GetCheckpointUtc().AsTime().Unix()
	}
	result.RCU = item.GetChangefeedSpec()
	if result.RCU == "" {
		result.RCU = request.GetChangefeedSpec()
	}
	result.TableConfig = nextGenTableConfig(nextGenTableConfigSource(request, item.GetChangefeedType()))
	result.StartPosition = nextGenStartPosition(nextGenStartPositionSource(request, item.GetChangefeedType()))
	config, err := sanitizeDownstreamConfig(nextGenDownstreamSource(request, item.GetChangefeedType()))
	if err != nil {
		return nil, err
	}
	result.DownstreamConfig = config
	return result, nil
}

func enrichDedicated(result *Changefeed, internal *dataflow.InternalChangefeed) {
	if result == nil || internal == nil {
		return
	}
	if internal.GetId() != 0 {
		result.ID = strconv.FormatUint(internal.GetId(), 10)
	}
	result.TiCDCCluster = dedicatedTiCDCCluster(internal.GetTicdcCluster())
	if result.TiCDCCluster != nil {
		result.TiCDCClusterName = result.TiCDCCluster.Name
	}
}

func enrichNextGen(result *Changefeed, internal *dataflowng.InternalChangefeed) {
	if result == nil || internal == nil {
		return
	}
	if internal.GetId() != 0 {
		result.ID = strconv.FormatUint(internal.GetId(), 10)
	}
	result.TiCDCCluster = nextGenTiCDCCluster(internal.GetTicdcCluster())
	if result.TiCDCCluster != nil {
		result.TiCDCClusterName = result.TiCDCCluster.Name
	}
}

func dedicatedTiCDCCluster(cluster *dataflow.InternalTiCDCCluster) *TiCDCCluster {
	if cluster == nil {
		return nil
	}
	createdAt := int64(0)
	if cluster.CreatedAt != nil {
		createdAt = cluster.CreatedAt.AsTime().Unix()
	}
	return &TiCDCCluster{
		ID:                strconv.FormatUint(cluster.GetId(), 10),
		Name:              cluster.GetName(),
		ClusterType:       "TICDC",
		TiDBClusterID:     strconv.FormatUint(cluster.GetTidbClusterId(), 10),
		Version:           cluster.GetVersion(),
		Region:            cluster.GetRegion(),
		ProviderType:      cluster.GetProviderType(),
		SpecificationCode: cluster.GetSpecificationCode(),
		Status:            cluster.GetStatus(),
		CreatedAt:         createdAt,
		DeployType:        cluster.GetDeployType(),
		ControlPlaneArch:  cluster.GetControlplaneArch(),
		OperatorVersion:   cluster.GetOperatorVersion(),
	}
}

func nextGenTiCDCCluster(cluster *dataflowng.InternalTiCDCCluster) *TiCDCCluster {
	if cluster == nil {
		return nil
	}
	createdAt := int64(0)
	if cluster.CreatedAt != nil {
		createdAt = cluster.CreatedAt.AsTime().Unix()
	}
	return &TiCDCCluster{
		ID:                strconv.FormatUint(cluster.GetId(), 10),
		Name:              cluster.GetName(),
		ClusterType:       "TICDC_NG",
		TiDBClusterID:     cluster.GetSharedPoolId(),
		Version:           cluster.GetVersion(),
		Region:            cluster.GetRegion(),
		ProviderType:      cluster.GetProviderType(),
		SpecificationCode: cluster.GetSpecificationCode(),
		Status:            cluster.GetStatus(),
		CreatedAt:         createdAt,
	}
}

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

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

func dedicatedTableConfig(input *dataflow.TableConfig) *TableConfig {
	if input == nil {
		return nil
	}
	result := &TableConfig{
		FilterRules: append([]string{}, input.GetFilterRule()...), Mode: normalizeEnum(input.GetMode().String(), "TableMode_"),
		CaseSensitive: input.GetCaseSensitive(), EventFilters: make([]*EventFilter, 0, len(input.GetEventFilterRule())),
	}
	for _, rule := range input.GetEventFilterRule() {
		if rule != nil {
			result.EventFilters = append(result.EventFilters, eventFilter(rule.GetMatcher(), rule.GetIgnoreEvent(), rule.GetIgnoreSql(),
				rule.GetIgnoreInsertValueExpr(), rule.GetIgnoreUpdateNewValueExpr(), rule.GetIgnoreUpdateOldValueExpr(), rule.GetIgnoreDeleteValueExpr()))
		}
	}
	return result
}

func nextGenTableConfig(input *dataflowng.TableConfig) *TableConfig {
	if input == nil {
		return nil
	}
	result := &TableConfig{
		FilterRules: append([]string{}, input.GetFilterRule()...), Mode: normalizeEnum(input.GetMode().String(), "TableMode_"),
		CaseSensitive: input.GetCaseSensitive(), EventFilters: make([]*EventFilter, 0, len(input.GetEventFilterRule())),
	}
	for _, rule := range input.GetEventFilterRule() {
		if rule != nil {
			result.EventFilters = append(result.EventFilters, eventFilter(rule.GetMatcher(), rule.GetIgnoreEvent(), rule.GetIgnoreSql(),
				rule.GetIgnoreInsertValueExpr(), rule.GetIgnoreUpdateNewValueExpr(), rule.GetIgnoreUpdateOldValueExpr(), rule.GetIgnoreDeleteValueExpr()))
		}
	}
	return result
}

func eventFilter(matchers, ignoredEvents, ignoredSQL []string, insert, updateNew, updateOld, deleteValue string) *EventFilter {
	return &EventFilter{
		TableMatchers: append([]string{}, matchers...), IgnoredEvents: append([]string{}, ignoredEvents...),
		IgnoredSQLStatements: append([]string{}, ignoredSQL...), IgnoredInsertValueExpression: insert,
		IgnoredUpdateNewValueExpression: updateNew, IgnoredUpdateOldValueExpression: updateOld,
		IgnoredDeleteValueExpression: deleteValue,
	}
}

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

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

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

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

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

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

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

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

func sanitizeDownstreamConfig(message proto.Message) (map[string]interface{}, error) {
	if message == nil || !message.ProtoReflect().IsValid() {
		return nil, nil
	}
	data, err := (protojson.MarshalOptions{UseProtoNames: true, EmitDefaultValues: true}).Marshal(message)
	if err != nil {
		return nil, err
	}
	var result map[string]interface{}
	if err := json.Unmarshal(data, &result); err != nil {
		return nil, err
	}
	sanitizeConfigMap(result)
	if len(result) == 0 {
		return nil, nil
	}
	return result, nil
}

func sanitizeConfigMap(config map[string]interface{}) {
	for key, value := range config {
		if key == "table_config" || key == "start_position" || sensitiveConfigKey(key) {
			delete(config, key)
			continue
		}
		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 sensitiveConfigKey(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
}

func dedicatedStatus(status dataflow.Connector_Status) string {
	switch status {
	case dataflow.Connector_RUNNING:
		return "normal"
	case dataflow.Connector_WARNING:
		return "warning"
	case dataflow.Connector_ERROR:
		return "error"
	case dataflow.Connector_PAUSING:
		return "stopping"
	case dataflow.Connector_PAUSED:
		return "stopped"
	case dataflow.Connector_RESUMING:
		return "starting"
	default:
		return strings.ToLower(status.String())
	}
}

func nextGenStatus(status dataflowng.Changefeed_Status) string {
	switch status {
	case dataflowng.Changefeed_RUNNING:
		return "normal"
	case dataflowng.Changefeed_WARNING:
		return "warning"
	case dataflowng.Changefeed_ERROR:
		return "error"
	case dataflowng.Changefeed_PAUSING:
		return "stopping"
	case dataflowng.Changefeed_PAUSED:
		return "stopped"
	case dataflowng.Changefeed_RESUMING:
		return "starting"
	default:
		return strings.ToLower(status.String())
	}
}

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

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

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