package domain

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

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

type ConfigComponent string

const (
	ConfigComponentTiCDC ConfigComponent = "ticdc"
)

type TiCDCConfigData struct {
	BaseImage *string                `json:"baseImage,omitempty"`
	Tag       *string                `json:"tag,omitempty"`
	Version   *string                `json:"version,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"`
}

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)
	}
	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"`
	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 (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.KafkaClientID != nil && *m.KafkaClientID != "" {
		uri += fmt.Sprintf("&kafka-client-id=%s", url.QueryEscape(*m.KafkaClientID))
	}
	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)
	}
	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"`
	// Endpoint is the custom S3-compatible storage endpoint (e.g. Alibaba
	// Cloud OSS). When non-empty it is appended to the sink URI as the
	// `endpoint` query parameter.
	Endpoint *string `json:"endpoint"`
	// Provider is the object-storage provider passed to TiCDC/BR as the
	// `provider` query parameter (e.g. "alibaba" for Alibaba Cloud OSS). When
	// non-empty it makes BR skip AWS-specific bucket-region discovery and use
	// path-style addressing. Empty means the default AWS S3 behavior.
	Provider *string `json:"provider"`
}

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.Endpoint != nil && *m.Endpoint != "" {
		uri += fmt.Sprintf("&endpoint=%s", url.QueryEscape(*m.Endpoint))
	}
	if m.Provider != nil && *m.Provider != "" {
		uri += fmt.Sprintf("&provider=%s", url.QueryEscape(*m.Provider))
	}
	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)
	}
	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"`
}

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)
	}
	return uri
}

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

type UpsertConfigTemplateReq struct {
	Name   string
	Data   string
	Enable bool
}

type UpsertConfigTemplateResp struct {
	ID uint64
}

type DisableConfigTemplateReq struct {
	Name string
}

type DisableConfigTemplateResp struct{}

type QueryConfigTemplateReq struct {
	Name string
}

type QueryConfigTemplateResp struct {
	ID   uint64
	Name string
	Data string
}

type GenerateConfigByTemplateReq struct {
	TemplateName string
	BusinessID   uint64
}

type GenerateConfigByTemplateResp struct {
	ID uint64
}

type UpsertConfigReq struct {
	BusinessID uint64
	Data       string
	Enable     bool
}

type UpsertConfigResp struct {
	ID uint64
}

type DeleteConfigReq struct {
	BusinessID uint64
}

type DeleteConfigResp struct{}

type QueryConfigReq struct {
	BusinessID uint64
}

type QueryConfigResp struct {
	ID         uint64
	BusinessID uint64
	Data       string
}

type CreateTiCDCClusterConfigReq struct {
	TiCDCClusterID     uint64
	TiDBClusterVersion string
	CloudProvider      entity.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 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
}

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

type UpdateTiCDCClusterStorageReq struct {
	TiCDCClusterID uint64
	SizeGi         uint32
	ThroughputMbs  uint32
	Iops           uint32
}