package central

import (
	"context"
	"crypto/tls"
	"database/sql"
	"encoding/json"
	"fmt"
	"github.com/pingcap/log"
	"github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl"
	"github.com/segmentio/kafka-go/sasl/plain"
	"github.com/segmentio/kafka-go/sasl/scram"
	"io/ioutil"
	"net"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"

	"github.com/tidbcloud/clusterproxy/resolve"
	"github.com/tidbcloud/clusterproxy/route"
	"github.com/tidbcloud/clusterproxy/route/router"
)

const MinInsyncReplicasConfigName = "min.insync.replicas"

func init() {
	router.RegisterRoute(testConnectionRoute)
}

var testConnectionRoute = &route.Route{
	Name:    "test-connection",
	Path:    "test_connection",
	Version: route.V1,
	Schema: &route.Schema{
		PathRegex:       "^$",
		RequiredHeaders: []string{},
		JSONBodyModel:   &connectionModel{},
	},
	GetHandler: connectionGetHandler,
}

type kafkaItem struct {
	ReplicationFactor       uint32 `json:"replication_factor"`
	PartitionNum            uint32 `json:"partition_num"`
	SchemaRegistryEndpoints string `json:"schema_registry_endpoints"`
}

type connectionModel struct {
	Type      string    `json:"type"`
	User      string    `json:"user"`
	Password  string    `json:"password"`
	Url       string    `json:"url"`
	Topic     string    `json:"topic"`
	AuthType  string    `json:"auth_type"`
	EnableTls bool      `json:"enable_tls"`
	KafkaItem kafkaItem `json:"kafka_item"`
}

func connectionGetHandler(c *route.Context, cl resolve.Cluster) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		body, err := ioutil.ReadAll(r.Body)
		defer r.Body.Close() // nolint
		if err != nil {
			http.Error(w, "failed to read body contents", http.StatusInternalServerError)
			return
		}

		model := &connectionModel{}
		err = json.Unmarshal(body, model)
		if err != nil {
			// Surface non-internal errors to the response.
			http.Error(w, fmt.Sprintf("failed to parse the body contents: %v", err), http.StatusBadRequest)
			return
		}
		log.Info(fmt.Sprintf("%v", model))

		var tlsCfg *tls.Config = nil
		if model.EnableTls {
			tlsCfg = &tls.Config{
				MinVersion: tls.VersionTLS12,
				NextProtos: []string{"h2", "http/1.1"},
			}
		}

		// ctx for dial kafka/mysql
		ctx, cancel := context.WithTimeout(context.TODO(), 3*time.Second)
		defer cancel()
		switch model.Type {
		case "kafka":
			if model.Url == "" {
				http.Error(w, "kafka url are required field", http.StatusBadRequest)
				return
			}
			var conn *kafka.Conn
			var client *kafka.Client
			switch model.AuthType {
			case "sasl-plain":
				mechanism := plain.Mechanism{Username: model.User, Password: model.Password}
				dialer := &kafka.Dialer{
					SASLMechanism: mechanism,
					TLS:           tlsCfg,
				}
				conn, err = dialer.Dial("tcp", model.Url)
				if err != nil {
					http.Error(w, fmt.Sprintf("unable to connect kafka authType %s url %s topic %s error %v", model.AuthType, model.Url, model.Topic, err), http.StatusInternalServerError)
					return
				}

				client, err = newKafkaClient(model.Url, tlsCfg, mechanism)
				if err != nil {
					http.Error(w, fmt.Sprintf("new kafka client failed %v", err), http.StatusInternalServerError)
					return
				}
			case "sasl-scram-sha-256":
				mechanism, err := scram.Mechanism(scram.SHA256, model.User, model.Password)
				if err != nil {
					http.Error(w, fmt.Sprintf("scram mechanism kafka authType %s url %s topic %s error %v", model.AuthType, model.Url, model.Topic, err), http.StatusInternalServerError)
					return
				}
				dialer := &kafka.Dialer{
					SASLMechanism: mechanism,
					TLS:           tlsCfg,
				}
				conn, err = dialer.Dial("tcp", model.Url)
				if err != nil {
					http.Error(w, fmt.Sprintf("unable to connect kafka authType %s url %s topic %s error %v", model.AuthType, model.Url, model.Topic, err), http.StatusInternalServerError)
					return
				}

				client, err = newKafkaClient(model.Url, tlsCfg, mechanism)
				if err != nil {
					http.Error(w, fmt.Sprintf("new kafka client failed %v", err), http.StatusInternalServerError)
					return
				}
			case "sasl-scram-sha-512":
				mechanism, err := scram.Mechanism(scram.SHA512, model.User, model.Password)
				if err != nil {
					http.Error(w, fmt.Sprintf("scram mechanism kafka authType %s url %s topic %s error %v", model.AuthType, model.Url, model.Topic, err), http.StatusInternalServerError)
					return
				}
				dialer := &kafka.Dialer{
					SASLMechanism: mechanism,
					TLS:           tlsCfg,
				}
				conn, err = dialer.Dial("tcp", model.Url)
				if err != nil {
					http.Error(w, fmt.Sprintf("unable to connect kafka authType %s url %s topic %s error %v", model.AuthType, model.Url, model.Topic, err), http.StatusInternalServerError)
					return
				}

				client, err = newKafkaClient(model.Url, tlsCfg, mechanism)
				if err != nil {
					http.Error(w, fmt.Sprintf("new kafka client failed %v", err), http.StatusInternalServerError)
					return
				}
			case "":
				dialer := &kafka.Dialer{
					TLS: tlsCfg,
				}
				conn, err = dialer.Dial("tcp", model.Url)
				if err != nil {
					http.Error(w, fmt.Sprintf("unable to connect kafka authType %s url %s topic %s error %v", model.AuthType, model.Url, model.Topic, err), http.StatusInternalServerError)
					return
				}

				client, err = newKafkaClient(model.Url, tlsCfg, nil)
				if err != nil {
					http.Error(w, fmt.Sprintf("new kafka client failed %v", err), http.StatusInternalServerError)
					return
				}
			default:
				http.Error(w, "unknown kafka auth type "+model.AuthType, http.StatusBadRequest)
				return
			}

			if conn == nil {
				http.Error(w, "dial kafka connect failed, empty conn", http.StatusBadRequest)
				return
			}
			brokers, err := conn.Brokers()
			if err != nil {
				http.Error(w, fmt.Sprintf("conn brokers error %v", err), http.StatusInternalServerError)
				return
			} else if len(brokers) == 0 {
				http.Error(w, fmt.Sprintf("broker number is 0"), http.StatusInternalServerError)
				return
			}

			if model.KafkaItem.ReplicationFactor != 0 {
				resp, err := client.DescribeConfigs(ctx, &kafka.DescribeConfigsRequest{
					Resources: []kafka.DescribeConfigRequestResource{
						{
							ResourceType: kafka.ResourceTypeBroker,
							ResourceName: strconv.Itoa(brokers[0].ID),
							ConfigNames:  []string{MinInsyncReplicasConfigName},
						},
						{
							ResourceType: kafka.ResourceTypeTopic,
							ResourceName: model.Topic,
							ConfigNames:  []string{MinInsyncReplicasConfigName},
						},
					},
				})
				log.Info(fmt.Sprintf("describe kafka configs resp: %+v, err: %+v", resp, err))
				if err == nil && len(resp.Resources) != 0 {
					for i := 0; i < len(resp.Resources); i++ {
						if len(resp.Resources[i].ConfigEntries) != 0 {
							minInsyncReplicasStr := resp.Resources[i].ConfigEntries[0].ConfigValue
							minInsyncReplicas, err := strconv.Atoi(minInsyncReplicasStr)
							if err != nil {
								http.Error(w, fmt.Sprintf("atoi min.insync.replicas config failed %v", err), http.StatusInternalServerError)
								return
							}
							log.Info(fmt.Sprintf("success connect kafka %s, minInsyncReplicas = %d, brokerNum = %d", model.Url, minInsyncReplicas, len(brokers)))
							if model.KafkaItem.ReplicationFactor < uint32(minInsyncReplicas) || model.KafkaItem.ReplicationFactor > uint32(len(brokers)) {
								http.Error(w, fmt.Sprintf("ReplicationFactor %d not in range [%s(%d),%s(%d)]",
									model.KafkaItem.ReplicationFactor, MinInsyncReplicasConfigName, minInsyncReplicas, "broker number", len(brokers)), http.StatusInternalServerError)
								return
							}
						}
					}
				}
			}
			if model.KafkaItem.PartitionNum != 0 {
				if model.KafkaItem.PartitionNum < 1 || model.KafkaItem.PartitionNum > 10*uint32(len(brokers)) {
					http.Error(w, fmt.Sprintf("PartitionNum %d not in range [%d,%d]",
						model.KafkaItem.PartitionNum, 1, 10*len(brokers)), http.StatusInternalServerError)
					return
				}
			}
			if model.KafkaItem.SchemaRegistryEndpoints != "" {
				schemaRegistry, err := url.Parse(model.KafkaItem.SchemaRegistryEndpoints)
				if err != nil {
					http.Error(w, fmt.Sprintf("schema registry endpoint parse url failed, %s", err.Error()), http.StatusInternalServerError)
					return
				}
				ipPortArr := strings.Split(schemaRegistry.Host, ":")
				if len(ipPortArr) != 2 {
					http.Error(w, "schema registry endpoint format invalid", http.StatusInternalServerError)
					return
				}
				schemaRegistryConn, err := net.DialTimeout("tcp", net.JoinHostPort(ipPortArr[0], ipPortArr[1]), 5*time.Second)
				if err != nil || schemaRegistryConn == nil {
					http.Error(w, "connect schema registry endpoint failed", http.StatusInternalServerError)
					return
				}
			}
			/*
				if model.Topic != "" {
					partitions, err := conn.ReadPartitions()
					if err != nil {
						http.Error(w, fmt.Sprintf("failed to fetch partitions"), http.StatusBadRequest)
						return
					}
					find := false
					for _, p := range partitions {
						if p.Topic == model.Topic {
							find = true
							break
						}
					}
					if !find {
						http.Error(w, fmt.Sprintf("Topic '%s' does not exist.", model.Topic), http.StatusBadRequest)
						return
					}
				}*/

			return
		case "mysql":
			if model.User == "" || model.Password == "" || model.Url == "" {
				http.Error(w, "user, password and url are required field", http.StatusBadRequest)
				return
			}
			db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@(%s)/", model.User, model.Password, model.Url))
			if err != nil {
				http.Error(w, fmt.Sprintf("unable to open sql connection url %s error %v", model.Url, err), http.StatusInternalServerError)
				return
			}
			if err := db.PingContext(ctx); err != nil {
				http.Error(w, fmt.Sprintf("unable to connect sql url %s error %v", model.Url, err), http.StatusInternalServerError)
				return
			}
		default:
			http.Error(w, fmt.Sprintf("invalid connection type %s", model.Type), http.StatusBadRequest)
			return
		}
	})
}

func newKafkaClient(url string, tls *tls.Config, sasl sasl.Mechanism) (*kafka.Client, error) {
	dialFunc := (&net.Dialer{}).DialContext
	transport := kafka.Transport{
		Dial: dialFunc,
	}
	if tls != nil {
		transport.TLS = tls
	}
	if sasl != nil {
		transport.SASL = sasl
	}

	client := &kafka.Client{
		Addr:      kafka.TCP(url),
		Timeout:   5 * time.Second,
		Transport: &transport,
	}

	return client, nil
}