package domain

import (
	"encoding/json"
	"fmt"
	"net/url"
	"regexp"

	"github.com/pingcap/tidb-operator/pkg/apis/util/k8s"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/pkgv2/clusterproxy"
)

type TiCDCConfigData struct {
	BaseImage *string                `json:"baseImage,omitempty"`
	Tag       *string                `json:"tag,omitempty"`
	Config    map[string]interface{} `json:"config,omitempty"`
}

func (t *TiCDCConfigData) UnmarshalJson(data string) error {
	if err := json.Unmarshal([]byte(data), t); err != nil {
		return err
	}
	configBytes, err := json.Marshal(t.Config)
	if err != nil {
		return err
	}
	t.Config = make(map[string]interface{})
	// In order to correctly deserialize the integer values of map types
	if err := k8s.Unmarshal(configBytes, &t.Config); err != nil {
		return err
	}
	return nil
}

type TiCDCChangefeedURIConfig interface {
	GenerateURI() string
}

type TiCDCChangefeedMySQLURIConfigData 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"`
	MultiStmtEnable       *bool   `json:"multi-stmt-enable"`
	MaxMultiUpdateRow     *int32  `json:"max-multi-update-row"`
	MaxMultiUpdateRowSize *int64  `json:"max-multi-update-row-size"`
	SslCa                 *string `json:"ssl-ca"`
	SslCert               *string `json:"ssl-cert"`
	SslKey                *string `json:"ssl-key"`
	ReadTimeout           *string `json:"read-timeout"`
	WriteTimeout          *string `json:"write-timeout"`
	Timeout               *string `json:"timeout"`
	TidbTxnMode           *string `json:"tidb-txn-mode"`
	SafeMode              *bool   `json:"safe-mode"`
}

//nolint:gocyclo
func (t *TiCDCChangefeedMySQLURIConfigData) GenerateURI() string {
	baseURI := fmt.Sprintf("mysql://%s:%s@%s?", *t.User, url.QueryEscape(*t.Password), *t.Endpoint)
	if t.MaxTxnRow != nil {
		baseURI = baseURI + fmt.Sprintf("max-txn-row=%d&", *t.MaxTxnRow)
	}
	if t.WorkerCount != nil {
		baseURI = baseURI + fmt.Sprintf("worker-count=%d&", *t.WorkerCount)
	}
	if t.CachePrepStmts != nil {
		baseURI = baseURI + fmt.Sprintf("cache-prep-stmts=%t&", *t.CachePrepStmts)
	}
	if t.TimeZone != nil {
		baseURI = baseURI + fmt.Sprintf("time-zone=%s&", *t.TimeZone)
	}
	if t.TransactionAtomicity != nil {
		baseURI = baseURI + fmt.Sprintf("transaction-atomicity=%s&", *t.TransactionAtomicity)
	}
	if t.BatchDmlEnable != nil {
		baseURI = baseURI + fmt.Sprintf("batch-dml-enable=%t&", *t.BatchDmlEnable)
	}
	if t.MultiStmtEnable != nil {
		baseURI = baseURI + fmt.Sprintf("multi-stmt-enable=%t&", *t.MultiStmtEnable)
	}
	if t.MaxMultiUpdateRow != nil {
		baseURI = baseURI + fmt.Sprintf("max-multi-update-row=%d&", *t.MaxMultiUpdateRow)
	}
	if t.MaxMultiUpdateRowSize != nil {
		baseURI = baseURI + fmt.Sprintf("max-multi-update-row-size=%d&", *t.MaxMultiUpdateRowSize)
	}
	if t.SslCa != nil {
		baseURI = baseURI + fmt.Sprintf("ssl-ca=%s&", *t.SslCa)
	}
	if t.SslCert != nil {
		baseURI = baseURI + fmt.Sprintf("ssl-cert=%s&", *t.SslCert)
	}
	if t.SslKey != nil {
		baseURI = baseURI + fmt.Sprintf("ssl-key=%s&", *t.SslKey)
	}
	if t.ReadTimeout != nil {
		baseURI = baseURI + fmt.Sprintf("read-timeout=%s&", *t.ReadTimeout)
	}
	if t.WriteTimeout != nil {
		baseURI = baseURI + fmt.Sprintf("write-timeout=%s&", *t.WriteTimeout)
	}
	if t.Timeout != nil {
		baseURI = baseURI + fmt.Sprintf("timeout=%s&", *t.Timeout)
	}
	if t.TidbTxnMode != nil {
		baseURI = baseURI + fmt.Sprintf("tidb-txn-mode=%s&", *t.TidbTxnMode)
	}
	if t.SafeMode != nil {
		baseURI = baseURI + fmt.Sprintf("safe-mode=%t&", *t.SafeMode)
	}
	re := regexp.MustCompile(`[?&]+$`)
	return re.ReplaceAllString(baseURI, "")
}

type TiCDCChangefeedKafkaURIConfigData 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"`
	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"`
	KafkaClientID                  *string `json:"kafka-client-id"`
	RequiredAcks                   *int32  `json:"required-acks"`
	MaxBatchSize                   *uint64 `json:"max-batch-size"`
	Ca                             *string `json:"ca"`
	Cert                           *string `json:"cert"`
	Key                            *string `json:"key"`
	SASLGssapiAuthType             *string `json:"sasl-gssapi-auth-type"`
	SASLGssapiKeytabPath           *string `json:"sasl-gssapi-keytab-path"`
	SASLGssapiKerberosConfigPath   *string `json:"sasl-gssapi-kerberos-config-path"`
	SASLGssapiServiceName          *string `json:"sasl-gssapi-service-name"`
	SASLGssapiUser                 *string `json:"sasl-gssapi-user"`
	SASLGssapiPassword             *string `json:"sasl-gssapi-password"`
	SASLGssapiRealm                *string `json:"sasl-gssapi-realm"`
	SASLGssapiDisablePafxfast      *bool   `json:"sasl-gssapi-disable-pafxfast"`
}

//nolint:funlen,gocyclo
func (m *TiCDCChangefeedKafkaURIConfigData) GenerateURI() string {
	uri := fmt.Sprintf(
		"kafka://%s/%s?protocol=%s&enable-tls=%t&replication-factor=%d&partition-num=%d&kafka-version=%s",
		m.BrokerEndpoints,
		m.DefaultTopic,
		m.Protocol,
		m.EnableTLS,
		m.ReplicationFactor,
		m.PartitionNum,
		m.KafkaVersion)
	if m.Compression != nil {
		uri += fmt.Sprintf("&compression=%s", *m.Compression)
	}
	if m.SASLMechanism != nil {
		uri += fmt.Sprintf("&sasl-mechanism=%s", *m.SASLMechanism)
	}
	if m.SASLUser != nil {
		uri += fmt.Sprintf("&sasl-user=%s", *m.SASLUser)
	}
	if m.SASLPassword != nil {
		uri += fmt.Sprintf("&sasl-password=%s", url.QueryEscape(*m.SASLPassword))
	}
	if m.AvroDecimalHandlingMode != nil {
		uri += fmt.Sprintf("&avro-decimal-handling-mode=%s", *m.AvroDecimalHandlingMode)
	}
	if m.AvroBigintUnsignedHandlingMode != nil {
		uri += fmt.Sprintf("&avro-bigint-unsigned-handling-mode=%s", *m.AvroBigintUnsignedHandlingMode)
	}
	if m.EnableTiDBExtension != nil {
		uri += fmt.Sprintf("&enable-tidb-extension=%t", *m.EnableTiDBExtension)
	}
	if m.AutoCreateTopic != nil {
		uri += fmt.Sprintf("&auto-create-topic=%t", *m.AutoCreateTopic)
	}
	if m.InsecureSkipVerify != nil {
		uri += fmt.Sprintf("&insecure-skip-verify=%t", *m.InsecureSkipVerify)
	}
	if m.MaxMessageBytes != nil {
		uri += fmt.Sprintf("&max-message-bytes=%d", *m.MaxMessageBytes)
	}
	if m.DialTimeout != nil {
		uri += fmt.Sprintf("&dial-timeout=%s", *m.DialTimeout)
	}
	if m.ReadTimeout != nil {
		uri += fmt.Sprintf("&read-timeout=%s", *m.ReadTimeout)
	}
	if m.WriteTimeout != nil {
		uri += fmt.Sprintf("&write-timeout=%s", *m.WriteTimeout)
	}
	if m.KafkaClientID != nil && *m.KafkaClientID != "" {
		uri += fmt.Sprintf("&kafka-client-id=%s", url.QueryEscape(*m.KafkaClientID))
	}
	if m.RequiredAcks != nil {
		uri += fmt.Sprintf("&required-acks=%d", *m.RequiredAcks)
	}
	if m.MaxBatchSize != nil {
		uri += fmt.Sprintf("&max-batch-size=%d", *m.MaxBatchSize)
	}
	if m.Ca != nil {
		uri += fmt.Sprintf("&ca=%s", *m.Ca)
	}
	if m.Cert != nil {
		uri += fmt.Sprintf("&cert=%s", *m.Cert)
	}
	if m.Key != nil {
		uri += fmt.Sprintf("&key=%s", *m.Key)
	}
	if m.SASLGssapiAuthType != nil {
		uri += fmt.Sprintf("&sasl-gssapi-auth-type=%s", *m.SASLGssapiAuthType)
	}
	if m.SASLGssapiKeytabPath != nil {
		uri += fmt.Sprintf("&sasl-gssapi-keytab-path=%s", *m.SASLGssapiKeytabPath)
	}
	if m.SASLGssapiKerberosConfigPath != nil {
		uri += fmt.Sprintf("&sasl-gssapi-kerberos-config-path=%s", *m.SASLGssapiKerberosConfigPath)
	}
	if m.SASLGssapiServiceName != nil {
		uri += fmt.Sprintf("&sasl-gssapi-service-name=%s", *m.SASLGssapiServiceName)
	}
	if m.SASLGssapiUser != nil {
		uri += fmt.Sprintf("&sasl-gssapi-user=%s", *m.SASLGssapiUser)
	}
	if m.SASLGssapiPassword != nil {
		uri += fmt.Sprintf("&sasl-gssapi-password=%s", url.QueryEscape(*m.SASLGssapiPassword))
	}
	if m.SASLGssapiRealm != nil {
		uri += fmt.Sprintf("&sasl-gssapi-realm=%s", *m.SASLGssapiRealm)
	}
	if m.SASLGssapiDisablePafxfast != nil {
		uri += fmt.Sprintf("&sasl-gssapi-disable-pafxfast=%t", *m.SASLGssapiDisablePafxfast)
	}
	return uri
}

type TiCDCChangefeedPulsarURIConfigData struct {
	PulsarProtocol      string `json:"pulsar-protocol"`
	Endpoint            string `json:"endpoint"`
	PulsarTenant        string `json:"pulsar-tenant"`
	PulsarNamespace     string `json:"pulsar-namespace"`
	DefaultTopic        string `json:"default-topic"`
	DataProtocol        string `json:"protocol"`
	EnableTiDBExtension *bool  `json:"enable-tidb-extension"`
}

func (m *TiCDCChangefeedPulsarURIConfigData) GenerateURI() string {
	uri := fmt.Sprintf("%s://%s/persistent://%s/%s/%s?protocol=%s",
		m.PulsarProtocol, m.Endpoint, m.PulsarTenant, m.PulsarNamespace, m.DefaultTopic, m.DataProtocol)
	if m.EnableTiDBExtension != nil {
		uri += fmt.Sprintf("&enable-tidb-extension=%t", *m.EnableTiDBExtension)
	}
	return uri
}

type TiCDCChangefeedS3URIConfigData 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"`
	WorkerCount          *uint32 `json:"worker-count"`
	Protocol             *string `json:"protocol"`
}

func (m *TiCDCChangefeedS3URIConfigData) GenerateURI() string {
	uri := fmt.Sprintf("%s?flush-interval=%s&file-size=%d",
		m.S3URI,
		m.FlushInterval,
		m.FileSize,
	)
	if m.RoleARN != nil && m.ExternalID != nil {
		uri += fmt.Sprintf("&role-arn=%s&external-id=%s", *m.RoleARN, *m.ExternalID)
	} else if m.AccessKey != nil && m.SecretAccessKey != nil {
		uri += fmt.Sprintf(
			"&access-key=%s&secret-access-key=%s", *m.AccessKey, url.QueryEscape(*m.SecretAccessKey))
	}
	if m.ForcePathStyle != nil {
		uri += fmt.Sprintf("&force-path-style=%t", *m.ForcePathStyle)
	}
	if m.TransactionAtomicity != nil {
		uri += fmt.Sprintf("&transaction-atomicity=%s", *m.TransactionAtomicity)
	}
	if m.EnableTiDBExtension != nil {
		uri += fmt.Sprintf("&enable-tidb-extension=%t", *m.EnableTiDBExtension)
	}
	if m.WorkerCount != nil {
		uri += fmt.Sprintf("&worker-count=%d", *m.WorkerCount)
	}
	if m.Protocol != nil {
		uri += fmt.Sprintf("&protocol=%s", *m.Protocol)
	}
	return uri
}

type TiCDCChangefeedGcsURIConfigData struct {
	GcsURI               string  `json:"gcs-uri"`
	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"`
	WorkerCount          *uint32 `json:"worker-count"`
	Protocol             *string `json:"protocol"`
}

func (m *TiCDCChangefeedGcsURIConfigData) GenerateURI() string {
	uri := fmt.Sprintf("%s?flush-interval=%s&file-size=%d",
		m.GcsURI,
		m.FlushInterval,
		m.FileSize,
	)
	if m.ForcePathStyle != nil {
		uri += fmt.Sprintf("&force-path-style=%t", *m.ForcePathStyle)
	}
	if m.TransactionAtomicity != nil {
		uri += fmt.Sprintf("&transaction-atomicity=%s", *m.TransactionAtomicity)
	}
	if m.EnableTiDBExtension != nil {
		uri += fmt.Sprintf("&enable-tidb-extension=%t", *m.EnableTiDBExtension)
	}
	if m.WorkerCount != nil {
		uri += fmt.Sprintf("&worker-count=%d", *m.WorkerCount)
	}
	if m.Protocol != nil {
		uri += fmt.Sprintf("&protocol=%s", *m.Protocol)
	}
	return uri
}

type TiCDCChangefeedAzureBlobStorageURIConfigData struct {
	BlobPath             string  `json:"blob-path"`
	AccountName          string  `json:"account-name"`
	SASToken             *string `json:"sas-token"`
	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"`
	WorkerCount          *uint32 `json:"worker-count"`
	Protocol             *string `json:"protocol"`
}

func (m *TiCDCChangefeedAzureBlobStorageURIConfigData) GenerateURI() string {
	uri := fmt.Sprintf("azure://%s?account-name=%s&flush-interval=%s&file-size=%d",
		m.BlobPath,
		m.AccountName,
		m.FlushInterval,
		m.FileSize,
	)
	if m.SASToken != nil {
		uri += fmt.Sprintf("&sas-token=%s", url.QueryEscape(*m.SASToken))
	}
	if m.ForcePathStyle != nil {
		uri += fmt.Sprintf("&force-path-style=%t", *m.ForcePathStyle)
	}
	if m.TransactionAtomicity != nil {
		uri += fmt.Sprintf("&transaction-atomicity=%s", *m.TransactionAtomicity)
	}
	if m.EnableTiDBExtension != nil {
		uri += fmt.Sprintf("&enable-tidb-extension=%t", *m.EnableTiDBExtension)
	}
	if m.WorkerCount != nil {
		uri += fmt.Sprintf("&worker-count=%d", *m.WorkerCount)
	}
	if m.Protocol != nil {
		uri += fmt.Sprintf("&protocol=%s", *m.Protocol)
	}
	return uri
}

type TiCDCChangefeedConfigData struct {
	StartTs       *uint64                     `json:"start_ts"`
	TargetTs      *uint64                     `json:"target_ts"`
	ReplicaConfig *clusterproxy.ReplicaConfig `json:"replica_config"`
}

type CreateTiCDCClusterConfigReq struct {
	TenantID           uint64
	TiCDCClusterID     uint64
	TiDBClusterVersion string
	CloudProvider      CloudProviderType
}

type CreateTiCDCClusterConfigResp struct {
	ConfigurationID uint64
}

type GetTiCDCClusterConfigReq struct {
	TiCDCClusterID uint64
}

type GetTiCDCClusterConfigResp struct {
	Configuration *TiCDCConfigData
}

type UpdateTiCDCClusterConfigReq struct {
	TiCDCClusterID     uint64
	TiDBClusterVersion *string
	TemplateID         *uint64
	CustomParams       map[string]*dataflow.ParamItem // the params will override the configuration
	CustomImageTag     *string
}

type PatchTiCDCClusterConfigReq struct {
	TiCDCClusterID uint64
	CustomParams   map[string]*dataflow.ParamItem // the params will be patched to the configuration
}

type GetNewTiCDCClusterImageReq struct {
	TiCDCClusterVersion string
}

type GetNewTiCDCClusterImageResp struct {
	BaseImage string
	Tag       string
}

type CreateChangefeedConfigReq struct {
	ChangefeedID   uint64
	DownstreamType entity.TiCDCChangefeedDownStreamType
}

type CreateChangefeedConfigResp struct {
	ChangefeedConfigurationID    uint64
	ChangefeedURIConfigurationID uint64
}

type GetChangefeedConfigReq struct {
	ChangefeedID uint64
}

type GetChangefeedConfigResp struct {
	Configuration    string
	URIConfiguration string
}

type UpdateChangefeedConfigReq struct {
	ChangefeedID     uint64
	Configuration    map[string]*dataflow.ParamItem
	URIConfiguration map[string]*dataflow.ParamItem
}