package controller

import (
	"crypto/tls"
	"encoding/json"
	"fmt"
	"github.com/pingcap/log"
	"github.com/pingcap/tep/pkg/local"
	"github.com/pingcap/tidb-operator/pkg/httputil"
	"github.com/pingcap/tidb-operator/pkg/pdapi"
	"go.uber.org/zap"
	"io/ioutil"
	"k8s.io/client-go/kubernetes"
	"net/http"
	"sync"
	"time"
)

const (
	DefaultTimeout = 2 * time.Second
)

// Namespace is a newtype of a string
type Namespace string

// PDControlInterface is an interface that knows how to manage and get tidb cluster's PD client
type PDControlInterface interface {
	// GetPDClient provides PDClient of the tidb cluster.
	GetPDClient(*local.PortForwarder, Namespace, string, bool) PDClient
}

// defaultPDControl is the default implementation of PDControlInterface.
type defaultPDControl struct {
	mutex     sync.Mutex
	kubeCli   kubernetes.Interface
	pdClients map[string]PDClient
}

// NewDefaultPDControl returns a defaultPDControl instance
func NewDefaultPDControl(kubeCli kubernetes.Interface) PDControlInterface {
	return &defaultPDControl{kubeCli: kubeCli, pdClients: map[string]PDClient{}}
}

func (c *Controller) GetPDClient(namespace Namespace, tcName string, tlsEnabled bool) PDClient {
	return c.PDCtl.GetPDClient(c.Forwarder, namespace, tcName, tlsEnabled)
}

// GetPDClient provides a PDClient of real pd cluster,if the PDClient not existing, it will create new one.
func (pdc *defaultPDControl) GetPDClient(portForwarder *local.PortForwarder, namespace Namespace, tcName string, tlsEnabled bool) PDClient {
	pdc.mutex.Lock()
	defer pdc.mutex.Unlock()

	var tlsConfig *tls.Config
	var err error
	scheme := "http"
	if tlsEnabled {
		scheme = "https"
	}
	key := pdClientKey(scheme, namespace, tcName)
	if _, ok := pdc.pdClients[key]; !ok {
		if tlsEnabled {
			tlsConfig, err = pdapi.GetTLSConfig(pdc.kubeCli, pdapi.Namespace(namespace), tcName, nil)
			if err != nil {
				log.Error(fmt.Sprintf("unable to get tls config for tidb cluster %q, pd client may not work", tcName), zap.Error(err))
				return &pdClient{url: pdClientURL(portForwarder, namespace, tcName, scheme), httpClient: &http.Client{Timeout: DefaultTimeout}}
			}
		}
		pdc.pdClients[key] = NewPDClient(pdClientURL(portForwarder, namespace, tcName, scheme), DefaultTimeout, tlsConfig)
	}
	return pdc.pdClients[key]
}

// pdClientKey returns the pd client key
func pdClientKey(scheme string, namespace Namespace, clusterName string) string {
	return fmt.Sprintf("%s.%s.%s", scheme, clusterName, string(namespace))
}

// pdClientUrl builds the url of pd client
func pdClientURL(portForwarder *local.PortForwarder, namespace Namespace, clusterName string, scheme string) string {
	if portForwarder != nil {
		localPort, err := portForwarder.ForwardServicePort(string(namespace), fmt.Sprintf("%s-pd", clusterName), 2379)
		if err != nil {
			log.Error("unable to forward", zap.Error(err))
		} else {
			return fmt.Sprintf("%s://localhost:%d", scheme, localPort)
		}
	}
	return fmt.Sprintf("%s://%s-pd.%s:2379", scheme, clusterName, string(namespace))
}

// PDClient provides pd server's api
type PDClient interface {
	// TransferPDLeader transfers pd leader to specified member
	TransferPDLeader(name string) error

	GetReplicate() (*Replicate, error)
}

var (
	replicatePrefix        = "pd/api/v1/config/replicate"
	pdLeaderTransferPrefix = "pd/api/v1/leader/transfer"
)

// pdClient is default implementation of PDClient
type pdClient struct {
	url        string
	httpClient *http.Client
}

// NewPDClient returns a new PDClient
func NewPDClient(url string, timeout time.Duration, tlsConfig *tls.Config) PDClient {
	return &pdClient{
		url: url,
		httpClient: &http.Client{
			Timeout:   timeout,
			Transport: &http.Transport{TLSClientConfig: tlsConfig},
		},
	}
}

type Replicate struct {
	MaxReplicas        uint32 `json:"max-replicas"`
	LocationLabels     string `json:"location-labels"`
	StrictlyMatchLabel string `json:"strictly-match-label"`
}

func (pc *pdClient) GetReplicate() (*Replicate, error) {
	apiURL := fmt.Sprintf("%s/%s", pc.url, replicatePrefix)
	body, err := httputil.GetBodyOK(pc.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	config := &Replicate{}
	err = json.Unmarshal(body, config)
	if err != nil {
		return nil, err
	}
	return config, nil
}

func (pc *pdClient) TransferPDLeader(memberName string) error {
	apiURL := fmt.Sprintf("%s/%s/%s", pc.url, pdLeaderTransferPrefix, memberName)
	req, err := http.NewRequest("POST", apiURL, nil)
	if err != nil {
		return err
	}
	res, err := pc.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer httputil.DeferClose(res.Body)
	if res.StatusCode == http.StatusOK || res.StatusCode == http.StatusNotFound {
		return nil
	}
	err2 := httputil.ReadErrorBody(res.Body)
	return fmt.Errorf("failed %v to transfer pd leader to %s,error: %v", res.StatusCode, memberName, err2)
}

func (pc *pdClient) getBodyOK(apiURL string) ([]byte, error) {
	res, err := pc.httpClient.Get(apiURL)
	if err != nil {
		return nil, err
	}
	defer httputil.DeferClose(res.Body)
	if res.StatusCode >= 400 {
		errMsg := fmt.Errorf(fmt.Sprintf("Error response %v URL %s", res.StatusCode, apiURL))
		return nil, errMsg
	}
	body, err := ioutil.ReadAll(res.Body)
	if err != nil {
		return nil, err
	}
	return body, err
}