// Copyright 2017 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package pdutil

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"net/http"
	"reflect"
	"strings"
	"time"

	"github.com/ngaut/log"
	"github.com/pingcap/kvproto/pkg/metapb"
	"github.com/pingcap/kvproto/pkg/pdpb"
	"github.com/pingcap/pd/pkg/typeutil"
	"github.com/pingcap/pd/server"
	"github.com/pingcap/tidb-operator/pkg/util/errors"
)

var (
	membersPrefix    = "pd/api/v1/members"
	storesPrefix     = "pd/api/v1/stores"
	storePrefix      = "pd/api/v1/store"
	configPrefix     = "pd/api/v1/config"
	clusterIDPrefix  = "pd/api/v1/cluster"
	schedulersPrefix = "pd/api/v1/schedulers"
)

// following struct definitions are copied from github.com/pingcap/pd/server/api/store
// these are not exported by that package

// MetaStore is TiKV store status defined in protobuf
type MetaStore struct {
	*metapb.Store
	StateName string `json:"state_name"`
}

// StoreStatus is TiKV store status returned from PD RESTful interface
type StoreStatus struct {
	StoreID            uint64            `json:"store_id"`
	Capacity           typeutil.ByteSize `json:"capacity"`
	Available          typeutil.ByteSize `json:"available"`
	LeaderCount        int               `json:"leader_count"`
	RegionCount        int               `json:"region_count"`
	SendingSnapCount   uint32            `json:"sending_snap_count"`
	ReceivingSnapCount uint32            `json:"receiving_snap_count"`
	ApplyingSnapCount  uint32            `json:"applying_snap_count"`
	IsBusy             bool              `json:"is_busy"`

	StartTS         time.Time         `json:"start_ts"`
	LastHeartbeatTS time.Time         `json:"last_heartbeat_ts"`
	Uptime          typeutil.Duration `json:"uptime"`
}

// StoreInfo is a single store info returned from PD RESTful interface
type StoreInfo struct {
	Store  *MetaStore   `json:"store"`
	Status *StoreStatus `json:"status"`
}

// StoresInfo is stores info returned from PD RESTful interface
type StoresInfo struct {
	Count  int          `json:"count"`
	Stores []*StoreInfo `json:"stores"`
}

// MembersInfo is PD members info returned from PD RESTful interface
//type Members map[string][]*pdpb.Member
type MembersInfo struct {
	Header     *pdpb.ResponseHeader `json:"header,omitempty"`
	Members    []*pdpb.Member       `json:"members,omitempty"`
	Leader     *pdpb.Member         `json:"leader,omitempty"`
	EtcdLeader *pdpb.Member         `json:"etcd_leader,omitempty"`
}

type schedulerInfo struct {
	Name    string `json:"name"`
	StoreID uint64 `json:"store_id"`
}

// Client provides information of pd client
type Client struct {
	url        string
	httpClient *http.Client
}

// New returns a Client
func New(url string, timeout time.Duration) *Client {
	return &Client{
		url:        url,
		httpClient: &http.Client{Timeout: timeout},
	}
}

// DeferClose captures the error returned from closing (if an error occurs).
// This is designed to be used in a defer statement.
func DeferClose(c io.Closer, err *error) {
	errors.DeferError(c.Close(), err)
}

// In the error case ready the body message.
// But return it as an error (or return an error from reading the body).
func readErrorBody(body io.Reader) (err error) {
	bodyBytes, err := ioutil.ReadAll(body)
	if err != nil {
		err = errors.Trace(err, 2)
	} else {
		err = errors.Errorf(string(bodyBytes))
	}
	return err
}

// GetConfig returns PD's config
func (cli *Client) GetConfig() (*server.Config, error) {
	apiURL := fmt.Sprintf("%s/%s", cli.url, configPrefix)
	body, err := cli.getBodyOK(apiURL)
	if err != nil {
		return nil, errors.Trace(err)
	}
	config := &server.Config{}
	err = json.Unmarshal(body, config)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return config, nil
}

func (cli *Client) getBodyOK(apiURL string) ([]byte, error) {
	res, err := cli.httpClient.Get(apiURL)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if res.StatusCode >= 400 {
		errMsg := fmt.Errorf(fmt.Sprintf("Error response %v", res.StatusCode))
		return nil, errors.Trace(errMsg)
	}

	defer DeferClose(res.Body, &err)
	body, err := ioutil.ReadAll(res.Body)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return body, err
}

// GetCluster is used when syncing pod labels.
func (cli *Client) GetCluster() (*metapb.Cluster, error) {
	apiURL := fmt.Sprintf("%s/%s", cli.url, clusterIDPrefix)
	body, err := cli.getBodyOK(apiURL)
	if err != nil {
		return nil, errors.Trace(err)
	}
	cluster := &metapb.Cluster{}
	err = json.Unmarshal(body, cluster)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return cluster, nil
}

// GetMembers returns all PD members from cluster
func (cli *Client) GetMembers() (*MembersInfo, error) {
	apiURL := fmt.Sprintf("%s/%s", cli.url, membersPrefix)
	body, err := cli.getBodyOK(apiURL)
	if err != nil {
		return nil, errors.Trace(err)
	}
	members := &MembersInfo{}
	err = json.Unmarshal(body, members)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return members, nil
}

// GetStores lists all TiKV stores from cluster
func (cli *Client) GetStores() (*StoresInfo, error) {
	apiURL := fmt.Sprintf("%s/%s", cli.url, storesPrefix)
	body, err := cli.getBodyOK(apiURL)
	if err != nil {
		return nil, err
	}
	log.Debugf("GetStores body: [%s]", string(body))
	storesInfo := &StoresInfo{}
	err = json.Unmarshal(body, storesInfo)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return storesInfo, nil
}

// GetTombStoneStores lists all tombstone stores from cluster
func (cli *Client) GetTombStoneStores() (*StoresInfo, error) {
	apiURL := fmt.Sprintf("%s/%s?state=2", cli.url, storesPrefix)
	body, err := cli.getBodyOK(apiURL)
	if err != nil {
		return nil, errors.Trace(err)
	}
	storesInfo := &StoresInfo{}
	err = json.Unmarshal(body, storesInfo)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return storesInfo, nil
}

// GetStore gets a TiKV store for a specific store id from cluster
func (cli *Client) GetStore(id uint64) (*StoreInfo, error) {
	apiURL := fmt.Sprintf("%s/%s/%d", cli.url, storePrefix, id)
	body, err := cli.getBodyOK(apiURL)
	if err != nil {
		return nil, errors.Trace(err)
	}
	storeInfo := &StoreInfo{}
	err = json.Unmarshal(body, storeInfo)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return storeInfo, nil
}

// DeleteStore deletes a TiKV store from cluster
func (cli *Client) DeleteStore(id string) error {
	apiURL := fmt.Sprintf("%s/%s/%s", cli.url, storePrefix, id)
	req, err := http.NewRequest("DELETE", apiURL, nil)
	if err != nil {
		return errors.Trace(err)
	}
	res, err := cli.httpClient.Do(req)
	if err != nil {
		return errors.Trace(err)
	}
	defer DeferClose(res.Body, &err)

	// Remove an offline store should return http.StatusOK
	if res.StatusCode == http.StatusOK {
		return nil
	}
	body, err := ioutil.ReadAll(res.Body)
	if err != nil {
		return errors.Trace(err)
	}

	// FIXME: We should not rely on error text
	// Remove a tombstone store should return "store has been removed"
	bodyStr := string(body)
	if strings.Contains(bodyStr, "store has been removed") {
		return nil
	}

	err = errors.Errorf("failed to delete store %s: %v", id, string(body))
	return err
}

// DeleteMemberByID delete a PD member from cluster
func (cli *Client) DeleteMemberByID(id string) error {
	apiURL := fmt.Sprintf("%s/%s/id/%s", cli.url, membersPrefix, id)
	req, err := http.NewRequest("DELETE", apiURL, nil)
	if err != nil {
		return errors.Trace(err)
	}
	res, err := cli.httpClient.Do(req)
	if err != nil {
		return errors.Trace(err)
	}
	defer DeferClose(res.Body, &err)
	if res.StatusCode == http.StatusOK {
		return nil
	}
	err2 := readErrorBody(res.Body)
	err = errors.Errorf("failed %v to delete member %s: %v", res.StatusCode, id, err2)
	return err
}

// DeleteMember delete a PD member from cluster
func (cli *Client) DeleteMember(name string) error {
	apiURL := fmt.Sprintf("%s/%s/name/%s", cli.url, membersPrefix, name)
	req, err := http.NewRequest("DELETE", apiURL, nil)
	if err != nil {
		return errors.Trace(err)
	}
	res, err := cli.httpClient.Do(req)
	if err != nil {
		return errors.Trace(err)
	}
	defer DeferClose(res.Body, &err)
	if res.StatusCode == http.StatusOK || res.StatusCode == http.StatusNotFound {
		return nil
	}
	err2 := readErrorBody(res.Body)
	err = errors.Errorf("failed %v to delete member %s: %v", res.StatusCode, name, err2)
	return err
}

// SetStoreLabels sets TiKV labels
func (cli *Client) SetStoreLabels(store *StoreInfo, labels map[string]string) (bool, error) {
	id := store.Store.GetId()
	if !storeLabelsEqualNodeLabels(store.Store.GetLabels(), labels) {
		log.Debugf("set store %d with labels: %v", id, labels)
		if err := cli.setStoreLabels(id, labels); err != nil {
			return false, errors.Trace(err)
		}
		return true, nil
	}
	return false, nil
}

func (cli *Client) setStoreLabels(id uint64, labels map[string]string) error {
	apiURL := fmt.Sprintf("%s/%s/%d/label", cli.url, storePrefix, id)
	data, err := json.Marshal(labels)
	if err != nil {
		return errors.Trace(err)
	}
	res, err := http.Post(apiURL, "application/json", bytes.NewBuffer(data))
	if err != nil {
		return errors.Trace(err)
	}
	defer DeferClose(res.Body, &err)
	if res.StatusCode == http.StatusOK {
		return nil
	}
	err2 := readErrorBody(res.Body)
	err = errors.Errorf("failed %v to set store labels: %v", res.StatusCode, err2)
	return err
}

// storeLabelsEqualNodeLabels compares store labels with node labels
// for historic reasons, PD stores TiKV labels as []*StoreLabel which is a key-value pair slice
func storeLabelsEqualNodeLabels(storeLabels []*metapb.StoreLabel, nodeLabels map[string]string) bool {
	ls := map[string]string{}
	for _, label := range storeLabels {
		key := label.GetKey()
		val := label.GetValue()
		ls[key] = val
	}
	return reflect.DeepEqual(ls, nodeLabels)
}

// BeginEvictLeader initiates leader eviction for a storeID.
// This is used when upgrading a pod.
func (cli *Client) BeginEvictLeader(storeID uint64) error {
	leaderEvictInfo := getLeaderEvictSchedulerInfo(storeID)
	apiURL := fmt.Sprintf("%s/%s", cli.url, schedulersPrefix)
	data, err := json.Marshal(leaderEvictInfo)
	if err != nil {
		return errors.Trace(err)
	}
	res, err := http.Post(apiURL, "application/json", bytes.NewBuffer(data))
	if err != nil {
		return errors.Trace(err)
	}
	defer DeferClose(res.Body, &err)
	if res.StatusCode == http.StatusOK {
		return nil
	}
	err2 := readErrorBody(res.Body)
	err = errors.Errorf("failed %v to begin evict leader of store:[%d],error: %v", res.StatusCode, storeID, err2)
	return err
}

// EndEvictLeader is used at the end of pod upgrade.
func (cli *Client) EndEvictLeader(storeID string) error {
	apiURL := fmt.Sprintf("%s/%s/%s", cli.url, schedulersPrefix, getLeaderEvictSchedulerStr(storeID))
	req, err := http.NewRequest("DELETE", apiURL, nil)
	if err != nil {
		return errors.Trace(err)
	}
	res, err := cli.httpClient.Do(req)
	if err != nil {
		return errors.Trace(err)
	}
	defer DeferClose(res.Body, &err)
	if res.StatusCode == http.StatusOK || res.StatusCode == http.StatusNotFound {
		return nil
	}
	err2 := readErrorBody(res.Body)
	err = errors.Errorf("failed %v to end leader evict scheduler of store [%s],error:%v", res.StatusCode, storeID, err2)
	return err
}

func getLeaderEvictSchedulerInfo(storeID uint64) *schedulerInfo {
	return &schedulerInfo{"evict-leader-scheduler", storeID}
}

func getLeaderEvictSchedulerStr(storeID string) string {
	return fmt.Sprintf("%s-%s", "evict-leader-scheduler", storeID)
}