package service

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

	"github.com/imdario/mergo"
	"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"
)

type changefeedURIConfig interface {
	GenerateURI() string
}

type changefeedMySQLURIConfig struct {
	Endpoint             *string `json:"endpoint"`
	User                 *string `json:"user"`
	Password             *string `json:"password"`
	MaxTxnRow            *int    `json:"max-txn-row"`
	WorkerCount          *int    `json:"worker-count"`
	CachePrepStmts       *bool   `json:"cache-prep-stmts"`
	TimeZone             *string `json:"time-zone"`
	TransactionAtomicity *string `json:"transaction-atomicity"`
	BatchDMLEnable       *bool   `json:"batch-dml-enable"`
}

func (c *changefeedMySQLURIConfig) GenerateURI() string {
	baseURI := fmt.Sprintf(
		"mysql://%s:%s@%s?",
		url.QueryEscape(value(c.User)),
		url.QueryEscape(value(c.Password)),
		value(c.Endpoint),
	)
	baseURI = appendIntQuery(baseURI, "max-txn-row", c.MaxTxnRow)
	baseURI = appendIntQuery(baseURI, "worker-count", c.WorkerCount)
	baseURI = appendBoolQuery(baseURI, "cache-prep-stmts", c.CachePrepStmts)
	baseURI = appendStringQuery(baseURI, "time-zone", c.TimeZone)
	baseURI = appendStringQuery(baseURI, "transaction-atomicity", c.TransactionAtomicity)
	baseURI = appendBoolQuery(baseURI, "batch-dml-enable", c.BatchDMLEnable)
	return strings.TrimRight(baseURI, "?&")
}

type changefeedKafkaURIConfig struct {
	BrokerEndpoints                string  `json:"broker-endpoints"`
	DefaultTopic                   string  `json:"default-topic"`
	Protocol                       string  `json:"protocol"`
	EnableTLS                      bool    `json:"enable-tls"`
	ReplicationFactor              uint32  `json:"replication-factor"`
	PartitionNum                   uint32  `json:"partition-num"`
	KafkaVersion                   string  `json:"kafka-version"`
	Compression                    *string `json:"compression"`
	KafkaClientID                  *string `json:"kafka-client-id"`
	SASLMechanism                  *string `json:"sasl-mechanism"`
	SASLUser                       *string `json:"sasl-user"`
	SASLPassword                   *string `json:"sasl-password"`
	AvroDecimalHandlingMode        *string `json:"avro-decimal-handling-mode"`
	AvroBigintUnsignedHandlingMode *string `json:"avro-bigint-unsigned-handling-mode"`
	EnableTiDBExtension            *bool   `json:"enable-tidb-extension"`
	AutoCreateTopic                *bool   `json:"auto-create-topic"`
	InsecureSkipVerify             *bool   `json:"insecure-skip-verify"`
	MaxMessageBytes                *uint64 `json:"max-message-bytes"`
	DialTimeout                    *string `json:"dial-timeout"`
	ReadTimeout                    *string `json:"read-timeout"`
	WriteTimeout                   *string `json:"write-timeout"`
}

func (c *changefeedKafkaURIConfig) GenerateURI() string {
	uri := fmt.Sprintf(
		"kafka://%s/%s?protocol=%s&enable-tls=%t&replication-factor=%d&partition-num=%d&kafka-version=%s",
		c.BrokerEndpoints, c.DefaultTopic, c.Protocol, c.EnableTLS,
		c.ReplicationFactor, c.PartitionNum, c.KafkaVersion,
	)
	// Every optional query parameter appender expects the URI to end with a
	// separator. Keep it independent of whether the first optional value exists.
	uri += "&"
	uri = appendStringQuery(uri, "compression", c.Compression)
	uri = appendStringQuery(uri, "kafka-client-id", c.KafkaClientID)
	uri = appendStringQuery(uri, "sasl-mechanism", c.SASLMechanism)
	uri = appendStringQuery(uri, "sasl-user", c.SASLUser)
	uri = appendStringQuery(uri, "sasl-password", c.SASLPassword)
	uri = appendStringQuery(uri, "avro-decimal-handling-mode", c.AvroDecimalHandlingMode)
	uri = appendStringQuery(uri, "avro-bigint-unsigned-handling-mode", c.AvroBigintUnsignedHandlingMode)
	uri = appendBoolQuery(uri, "enable-tidb-extension", c.EnableTiDBExtension)
	uri = appendBoolQuery(uri, "auto-create-topic", c.AutoCreateTopic)
	uri = appendBoolQuery(uri, "insecure-skip-verify", c.InsecureSkipVerify)
	if c.MaxMessageBytes != nil {
		uri += fmt.Sprintf("max-message-bytes=%d&", *c.MaxMessageBytes)
	}
	uri = appendStringQuery(uri, "dial-timeout", c.DialTimeout)
	uri = appendStringQuery(uri, "read-timeout", c.ReadTimeout)
	uri = appendStringQuery(uri, "write-timeout", c.WriteTimeout)
	return strings.TrimSuffix(uri, "&")
}

type changefeedS3URIConfig struct {
	S3URI                string  `json:"s3-uri"`
	AccessKey            *string `json:"access-key"`
	SecretAccessKey      *string `json:"secret-access-key"`
	RoleARN              *string `json:"role-arn"`
	ExternalID           *string `json:"external-id"`
	ForcePathStyle       *bool   `json:"force-path-style"`
	FlushInterval        string  `json:"flush-interval"`
	FileSize             uint64  `json:"file-size"`
	TransactionAtomicity *string `json:"transaction-atomicity"`
	EnableTiDBExtension  *bool   `json:"enable-tidb-extension"`
	Endpoint             *string `json:"endpoint"`
	Provider             *string `json:"provider"`
}

func (c *changefeedS3URIConfig) GenerateURI() string {
	uri := fmt.Sprintf("%s?flush-interval=%s&file-size=%d&", c.S3URI, c.FlushInterval, c.FileSize)
	uri = appendStringQuery(uri, "role-arn", c.RoleARN)
	uri = appendStringQuery(uri, "external-id", c.ExternalID)
	uri = appendStringQuery(uri, "access-key", c.AccessKey)
	uri = appendStringQuery(uri, "secret-access-key", c.SecretAccessKey)
	uri = appendStringQuery(uri, "endpoint", c.Endpoint)
	uri = appendStringQuery(uri, "provider", c.Provider)
	uri = appendBoolQuery(uri, "force-path-style", c.ForcePathStyle)
	uri = appendStringQuery(uri, "transaction-atomicity", c.TransactionAtomicity)
	uri = appendBoolQuery(uri, "enable-tidb-extension", c.EnableTiDBExtension)
	return strings.TrimSuffix(uri, "&")
}

type changefeedGCSURIConfig struct {
	GCSURI               string  `json:"gcs-uri"`
	Credentials          *string `json:"credentials"`
	ForcePathStyle       *bool   `json:"force-path-style"`
	FlushInterval        string  `json:"flush-interval"`
	FileSize             uint64  `json:"file-size"`
	TransactionAtomicity *string `json:"transaction-atomicity"`
	EnableTiDBExtension  *bool   `json:"enable-tidb-extension"`
}

func (c *changefeedGCSURIConfig) GenerateURI() string {
	uri := fmt.Sprintf("%s?flush-interval=%s&file-size=%d&", c.GCSURI, c.FlushInterval, c.FileSize)
	uri = appendStringQuery(uri, "credentials", c.Credentials)
	uri = appendBoolQuery(uri, "force-path-style", c.ForcePathStyle)
	uri = appendStringQuery(uri, "transaction-atomicity", c.TransactionAtomicity)
	uri = appendBoolQuery(uri, "enable-tidb-extension", c.EnableTiDBExtension)
	return strings.TrimSuffix(uri, "&")
}

func assembleChangefeedSinkURI(
	changefeed *entity.CDCChangefeed, userConfig *serverless.ChangefeedConfig, configCenterText string,
) (string, bizerr.BizError) {
	if userConfig == nil || userConfig.SinkURI == nil || userConfig.ReplicaConfig == nil {
		return "", bizerr.New(bizerr.ErrCodeParams, "user changefeed sink configuration is required")
	}
	parsed, err := url.Parse(*userConfig.SinkURI)
	if err != nil {
		return "", bizerr.WrapError(err, "parse user changefeed sink uri")
	}

	var userDefined, configCenterDefined changefeedURIConfig
	switch changefeed.SinkType {
	case constant.ChangefeedTypeKafka:
		// Essential stores Kafka sink settings in structured fields. The raw URI
		// only carries broker endpoints and the topic.
		userDefined = kafkaURIConfigFromUser(parsed, userConfig.ReplicaConfig)
		configCenterDefined = new(changefeedKafkaURIConfig)
	case constant.ChangefeedTypeMySQL:
		// Essential stores MySQL sink settings in structured fields. The raw URI
		// only carries the endpoint and credentials.
		userDefined = mysqlURIConfigFromUser(parsed, userConfig.ReplicaConfig)
		configCenterDefined = new(changefeedMySQLURIConfig)
	case constant.ChangefeedTypeCloudStorage:
		storageType, typeErr := changefeedDownstreamType(changefeed)
		if typeErr != nil {
			return "", typeErr
		}
		switch storageType {
		case "s3":
			var uriErr bizerr.BizError
			userDefined, uriErr = s3URIConfigFromUser(parsed, userConfig.ReplicaConfig)
			if uriErr != nil {
				return "", uriErr
			}
			configCenterDefined = new(changefeedS3URIConfig)
		case "gcs":
			var uriErr bizerr.BizError
			userDefined, uriErr = gcsURIConfigFromUser(parsed, userConfig.ReplicaConfig)
			if uriErr != nil {
				return "", uriErr
			}
			configCenterDefined = new(changefeedGCSURIConfig)
		default:
			return mergeGenericSinkURI(parsed, configCenterText)
		}
	default:
		return "", bizerr.New(bizerr.ErrCodeParams, "unsupported changefeed sink type: %s", changefeed.SinkType)
	}

	decoder := json.NewDecoder(strings.NewReader(configCenterText))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(configCenterDefined); err != nil {
		return "", bizerr.WrapError(err, "unmarshal changefeed uri config")
	}
	if err := mergo.Merge(
		configCenterDefined, userDefined, mergo.WithOverride, mergo.WithTransformers(&boolMergoTransformer{}),
	); err != nil {
		return "", bizerr.WrapError(err, "override changefeed uri config")
	}
	return configCenterDefined.GenerateURI(), nil
}

func kafkaURIConfigFromUser(uri *url.URL, replica *serverless.ReplicaConfig) *changefeedKafkaURIConfig {
	result := &changefeedKafkaURIConfig{
		BrokerEndpoints: uri.Host,
		DefaultTopic:    strings.TrimPrefix(uri.Path, "/"),
	}
	if replica.Sink == nil {
		return result
	}
	result.Protocol = value(replica.Sink.Protocol)
	if kafka := replica.Sink.KafkaConfig; kafka != nil {
		result.EnableTLS = boolValue(kafka.EnableTLS)
		result.ReplicationFactor = uint32(int16Value(kafka.ReplicationFactor))
		result.PartitionNum = uint32(int32Value(kafka.PartitionNum))
		result.KafkaVersion = value(kafka.KafkaVersion)
		result.Compression = kafka.Compression
		result.KafkaClientID = kafka.KafkaClientID
		result.SASLMechanism = kafka.SASLMechanism
		result.SASLUser = kafka.SASLUser
		result.SASLPassword = kafka.SASLPassword
		result.AutoCreateTopic = kafka.AutoCreateTopic
		result.InsecureSkipVerify = kafka.InsecureSkipVerify
		result.DialTimeout = kafka.DialTimeout
		result.ReadTimeout = kafka.ReadTimeout
		result.WriteTimeout = kafka.WriteTimeout
		if kafka.MaxMessageBytes != nil {
			v := uint64(*kafka.MaxMessageBytes)
			result.MaxMessageBytes = &v
		}
		if kafka.CodecConfig != nil {
			result.EnableTiDBExtension = kafka.CodecConfig.EnableTidbExtension
			result.AvroDecimalHandlingMode = kafka.CodecConfig.AvroDecimalHandlingMode
			result.AvroBigintUnsignedHandlingMode = kafka.CodecConfig.AvroBigintUnsignedHandlingMode
		}
	}
	return result
}

func mysqlURIConfigFromUser(uri *url.URL, replica *serverless.ReplicaConfig) *changefeedMySQLURIConfig {
	result := &changefeedMySQLURIConfig{Endpoint: stringPointer(uri.Host)}
	if uri.User != nil {
		result.User = stringPointer(uri.User.Username())
		if password, ok := uri.User.Password(); ok {
			result.Password = stringPointer(password)
		}
	}
	if replica.Sink == nil || replica.Sink.MySQLConfig == nil {
		return result
	}
	mysql := replica.Sink.MySQLConfig
	result.MaxTxnRow = mysql.MaxTxnRow
	result.WorkerCount = mysql.WorkerCount
	result.CachePrepStmts = mysql.EnableCachePreparedStatement
	result.TimeZone = mysql.TimeZone
	result.BatchDMLEnable = mysql.EnableBatchDML
	return result
}

func s3URIConfigFromUser(
	uri *url.URL, replica *serverless.ReplicaConfig,
) (*changefeedS3URIConfig, bizerr.BizError) {
	baseURI, query := splitSinkURI(uri)
	forcePathStyle, err := queryBoolPointer(query, "force-path-style")
	if err != nil {
		return nil, err
	}
	enableTiDBExtension, err := queryBoolPointer(query, "enable-tidb-extension")
	if err != nil {
		return nil, err
	}
	result := &changefeedS3URIConfig{
		S3URI:                baseURI,
		AccessKey:            queryStringPointer(query, "access-key"),
		SecretAccessKey:      queryStringPointer(query, "secret-access-key"),
		RoleARN:              queryStringPointer(query, "role-arn"),
		ExternalID:           queryStringPointer(query, "external-id"),
		ForcePathStyle:       forcePathStyle,
		TransactionAtomicity: queryStringPointer(query, "transaction-atomicity"),
		EnableTiDBExtension:  enableTiDBExtension,
		Endpoint:             queryStringPointer(query, "endpoint"),
		Provider:             queryStringPointer(query, "provider"),
	}
	fillCloudStorageURIConfig(replica, &result.FlushInterval, &result.FileSize)
	return result, nil
}

func gcsURIConfigFromUser(
	uri *url.URL, replica *serverless.ReplicaConfig,
) (*changefeedGCSURIConfig, bizerr.BizError) {
	baseURI, query := splitSinkURI(uri)
	forcePathStyle, err := queryBoolPointer(query, "force-path-style")
	if err != nil {
		return nil, err
	}
	enableTiDBExtension, err := queryBoolPointer(query, "enable-tidb-extension")
	if err != nil {
		return nil, err
	}
	result := &changefeedGCSURIConfig{
		GCSURI:               baseURI,
		Credentials:          queryStringPointer(query, "credentials"),
		ForcePathStyle:       forcePathStyle,
		TransactionAtomicity: queryStringPointer(query, "transaction-atomicity"),
		EnableTiDBExtension:  enableTiDBExtension,
	}
	fillCloudStorageURIConfig(replica, &result.FlushInterval, &result.FileSize)
	return result, nil
}

func fillCloudStorageURIConfig(replica *serverless.ReplicaConfig, flushInterval *string, fileSize *uint64) {
	if replica.Sink == nil || replica.Sink.CloudStorageConfig == nil {
		return
	}
	cloud := replica.Sink.CloudStorageConfig
	if cloud.FlushInterval != nil {
		*flushInterval = *cloud.FlushInterval
	}
	if cloud.FileSize != nil {
		*fileSize = uint64(*cloud.FileSize)
	}
}

func mergeGenericSinkURI(userURI *url.URL, configCenterText string) (string, bizerr.BizError) {
	configCenterDefined := map[string]any{}
	if err := json.Unmarshal([]byte(configCenterText), &configCenterDefined); err != nil {
		return "", bizerr.WrapError(err, "unmarshal changefeed uri config")
	}
	query := userURI.Query()
	for key, raw := range configCenterDefined {
		if query.Has(key) {
			continue
		}
		switch typed := raw.(type) {
		case string:
			query.Set(key, typed)
		case bool:
			query.Set(key, strconv.FormatBool(typed))
		case float64:
			query.Set(key, strconv.FormatFloat(typed, 'f', -1, 64))
		default:
			return "", bizerr.New(bizerr.ErrCodeParams, "unsupported uri config value for %s", key)
		}
	}
	userURI.RawQuery = query.Encode()
	return userURI.String(), nil
}

func splitSinkURI(uri *url.URL) (string, url.Values) {
	query := uri.Query()
	clone := *uri
	clone.RawQuery = ""
	clone.ForceQuery = false
	return clone.String(), query
}

func queryStringPointer(query url.Values, key string) *string {
	if !query.Has(key) {
		return nil
	}
	return stringPointer(query.Get(key))
}

func queryBoolPointer(query url.Values, key string) (*bool, bizerr.BizError) {
	if !query.Has(key) {
		return nil, nil
	}
	value, err := strconv.ParseBool(query.Get(key))
	if err != nil {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"invalid boolean sink uri parameter %s: %s",
			key,
			query.Get(key),
		)
	}
	return &value, nil
}

func stringPointer(v string) *string { return &v }

func value(v *string) string {
	if v == nil {
		return ""
	}
	return *v
}

func boolValue(v *bool) bool { return v != nil && *v }

func int16Value(v *int16) int16 {
	if v == nil {
		return 0
	}
	return *v
}

func int32Value(v *int32) int32 {
	if v == nil {
		return 0
	}
	return *v
}

func appendStringQuery(uri, key string, item *string) string {
	if item != nil {
		uri += fmt.Sprintf("%s=%s&", key, url.QueryEscape(*item))
	}
	return uri
}

func appendIntQuery(uri, key string, item *int) string {
	if item != nil {
		uri += fmt.Sprintf("%s=%d&", key, *item)
	}
	return uri
}

func appendBoolQuery(uri, key string, item *bool) string {
	if item != nil {
		uri += fmt.Sprintf("%s=%t&", key, *item)
	}
	return uri
}