// 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 spec

package framework

import (
	"database/sql"
	"encoding/json"
	serrors "errors"
	"fmt"
	"net/http"
	"net/url"
	"strings"

	_ "github.com/go-sql-driver/mysql"
	"github.com/juju/errors"
	"github.com/pingcap/tidb-cloud-manager/pkg/client"
	"github.com/pingcap/tidb-cloud-manager/pkg/manager"

	tictl "github.com/pingcap/tidb-cloud-manager/pkg/tidbinstance/controller"
	tsctl "github.com/pingcap/private-tidb-operator/pkg/tidbset/controller"
	"github.com/pingcap/private-tidb-operator/pkg/util/label"
	batchv1 "k8s.io/api/batch/v1"
	"k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
)

var (
	ClusterNotFound   = serrors.New("TidbCluster is not found")
	JobFailed         = serrors.New("Job has failed")
	PodsSizeNotEqual  = serrors.New("Not all of pods's size are expected")
	InitNotCompleted  = serrors.New("TiDB initialization hasn't completed")
	PodsNotAllRunning = serrors.New("Not all of pods are running")
)

const (
	rootPath       = "/pingcap.com/api/v1/clusters"
	configPath     = "/tmp/data/e2e_config.json"
	contentType    = "application/json"
	restoreImage   = "uhub.ucloud.cn/pingcap/tidb-enterprise-tools"
	RestoreJobName = "restore-data"
)

//clusterManager operates tidbcluster by cloud-manager's APIs
type ClusterManager struct {
	*HttpClient
	Endpoint    string
	ClusterName string
	Cli         client.Interface
}

func NewClusterManger(clusterName string) (*ClusterManager, error) {
	cli, err := client.New("")
	if err != nil {
		return nil, err
	}

	return &ClusterManager{
		Endpoint:    cloudManagerUrl,
		ClusterName: clusterName,
		HttpClient:  &HttpClient{http.DefaultClient},
		Cli:         cli,
	}, nil
}

func (cm *ClusterManager) getUrl() string {
	return fmt.Sprintf("%s%s", strings.TrimRight(cm.Endpoint, "/"), rootPath)
}

func (cm *ClusterManager) generateUrl(suffix string) string {
	return fmt.Sprintf("%s/%s", cm.getUrl(), strings.TrimLeft(suffix, "/"))
}

func (cm *ClusterManager) CreateTidbCluster(tc *manager.Cluster) error {
	tcBytes, err := json.Marshal(tc)
	if err != nil {
		return errors.Trace(err)
	}
	resp, err := cm.HttpClient.DoPost(cm.getUrl(), contentType, tcBytes)
	if err != nil {
		return errors.Trace(err)
	}
	if resp.StatusCode != http.StatusOK {
		return errors.Trace(fmt.Errorf(resp.Message))
	}
	return nil
}

func (cm *ClusterManager) ListTidbClusters() ([]manager.Cluster, error) {
	resp, err := cm.HttpClient.DoGet(cm.getUrl())
	if err != nil {
		return nil, errors.Trace(err)
	}
	if resp.StatusCode != http.StatusOK {
		return nil, errors.Trace(fmt.Errorf(resp.Message))
	}
	var clusters []manager.Cluster
	for _, c := range resp.Payload.Clusters {
		clusters = append(clusters, *c)
	}
	return clusters, nil
}

func (cm *ClusterManager) GetTidbCluster() (*manager.Cluster, error) {
	resp, err := cm.HttpClient.DoGet(cm.generateUrl(cm.ClusterName))
	if err != nil {
		return nil, errors.Trace(err)
	}
	if resp.StatusCode != http.StatusOK {
		return nil, errors.Trace(fmt.Errorf(resp.Message))
	}
	if len(resp.Payload.Clusters) == 0 {
		return nil, ClusterNotFound
	}
	return resp.Payload.Clusters[0], nil
}

func (cm *ClusterManager) ResetTidbPassword(passwd string) (string, error) {
	urlSuffix := fmt.Sprintf("%s/password", cm.ClusterName)
	passwdBytes, _ := json.Marshal(map[string]string{
		"root_password": passwd,
	})
	resp, err := cm.HttpClient.DoPut(cm.generateUrl(urlSuffix), contentType, passwdBytes)
	if err != nil {
		return "", errors.Trace(err)
	}
	if resp.StatusCode != http.StatusOK {
		return "", errors.Trace(fmt.Errorf(resp.Message))
	}
	if resp.Payload.Job == nil {
		return "", errors.Trace(fmt.Errorf("response job status is invalid"))
	}
	return resp.Payload.Job.Name, nil
}

func (cm *ClusterManager) CheckTidbConn(password string) error {
	dbDSN := fmt.Sprintf("root:%s@tcp(tidb-cluster-tidb.%s:4000)/?charset=utf8", password, cm.ClusterName)
	db, err := sql.Open("mysql", dbDSN)
	if err != nil {
		return errors.Trace(err)
	}
	defer db.Close()

	err = db.Ping()
	if err != nil {
		Errorf("Validate Tidb's password [%s] failed", password)
		return errors.Trace(err)
	}
	return nil
}

func (cm *ClusterManager) GetJobStatus(jobName string) (manager.JobConditionType, error) {
	status := manager.JobUnknown
	urlSuffix := fmt.Sprintf("%s/jobs/%s", cm.ClusterName, jobName)
	resp, err := cm.HttpClient.DoGet(cm.generateUrl(urlSuffix))
	if err != nil {
		return status, errors.Trace(err)
	}
	if resp.StatusCode != http.StatusOK {
		return status, errors.Trace(fmt.Errorf(resp.Message))
	}
	if resp.Payload.Job == nil {
		return status, errors.Trace(fmt.Errorf("get job status failed"))
	}
	status = resp.Payload.Job.Status
	return status, nil
}

func (cm *ClusterManager) UpdateTidbCluster(tc *manager.Cluster) error {
	tcBytes, err := json.Marshal(tc)
	if err != nil {
		return errors.Trace(err)
	}
	resp, err := cm.HttpClient.DoPut(cm.generateUrl(cm.ClusterName), contentType, tcBytes)
	if err != nil {
		return errors.Trace(err)
	}
	if resp.StatusCode != http.StatusOK {
		return errors.Trace(fmt.Errorf(resp.Message))
	}
	return nil
}

func (cm *ClusterManager) DeleteTidbCluster() error {
	resp, err := cm.HttpClient.DoDelete(cm.generateUrl(cm.ClusterName))
	if err != nil {
		return errors.Trace(err)
	}
	if resp.StatusCode != http.StatusOK {
		return errors.Trace(fmt.Errorf(resp.Message))
	}
	return nil
}

func (cm *ClusterManager) RestoreTidbCluster() error {
	url := fmt.Sprintf("%s/restore", cm.generateUrl(cm.ClusterName))
	resp, err := cm.HttpClient.DoPut(url, contentType, nil)
	if err != nil {
		return errors.Trace(err)
	}
	if resp.StatusCode != http.StatusOK {
		return errors.Trace(fmt.Errorf(resp.Message))
	}
	return nil
}

func (cm *ClusterManager) DeletePod(podName string) error {
	urlSuffix := fmt.Sprintf("%s/pods/%s", cm.ClusterName, podName)
	resp, err := cm.HttpClient.DoDelete(cm.generateUrl(urlSuffix))
	if err != nil {
		return errors.Trace(err)
	}
	if resp.StatusCode != http.StatusOK {
		return errors.Trace(fmt.Errorf(resp.Message))
	}
	return nil
}

func (cm *ClusterManager) CreateBackup(db string, tables []string, nodeSelector map[string]string) (string, error) {
	urlSuffix := fmt.Sprintf("%s/backup", cm.ClusterName)
	param := manager.BackupParam{
		ImageVersion: "latest",
		Database:     db,
		Tables:       tables,
		NodeSelector: nodeSelector,
	}
	body, err := json.Marshal(param)
	if err != nil {
		return "", errors.Trace(err)
	}
	resp, err := cm.HttpClient.DoPost(cm.generateUrl(urlSuffix), contentType, body)
	if err != nil {
		return "", errors.Trace(err)
	}
	if resp.StatusCode != http.StatusOK {
		return "", errors.Trace(fmt.Errorf(resp.Message))
	}
	return resp.Payload.Job.Name, nil
}

func (cm *ClusterManager) GetBackup(name string) (*manager.BackupStatus, error) {
	urlSuffix := fmt.Sprintf("%s/backup/%s", cm.ClusterName, name)
	resp, err := cm.HttpClient.DoGet(cm.generateUrl(urlSuffix))
	if err != nil {
		return nil, errors.Trace(err)
	}
	if resp.StatusCode != http.StatusOK {
		return nil, errors.Trace(fmt.Errorf(resp.Message))
	}
	return resp.Payload.BackupStatus, nil
}

func (cm *ClusterManager) GetVolumes(ns, nodeName string, canSafeDeleted bool) ([]manager.Volume, error) {
	values := url.Values{}
	values.Set("namespace", ns)
	values.Set("node_name", nodeName)
	if canSafeDeleted {
		values.Set("can_safe_deleted", "true")
	}
	urlSuffix := fmt.Sprintf("/pingcap.com/api/v1/volumes?%s", values.Encode())
	fullURL := strings.TrimRight(cm.Endpoint, "/") + urlSuffix
	resp, err := cm.HttpClient.DoGet(fullURL)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if resp.StatusCode != http.StatusOK {
		return nil, errors.Trace(fmt.Errorf(resp.Message))
	}

	vols := []manager.Volume{}
	for _, vol := range resp.Payload.Volumes {
		vols = append(vols, vol)
	}
	return vols, nil
}

func (cm *ClusterManager) DeleteVolume(volName string) error {
	urlSuffix := fmt.Sprintf("/pingcap.com/api/v1/volumes/%s", volName)
	fullURL := strings.TrimRight(cm.Endpoint, "/") + urlSuffix
	resp, err := cm.HttpClient.DoDelete(fullURL)
	if err != nil {
		return errors.Trace(errors.Errorf("fullURL: %s, %v", fullURL, err))
	}
	if resp.StatusCode != http.StatusOK {
		return errors.Trace(fmt.Errorf(resp.Message))
	}

	return nil
}

func (cm *ClusterManager) CheckClusterStatus(tc, rtc *manager.Cluster) error {
	//first check the pod's size
	if len(rtc.PdStatus) != tc.Pd.Size || len(rtc.TidbStatus) != tc.Tidb.Size || len(rtc.TikvStatus) != tc.Tikv.Size {
		return errors.Trace(PodsSizeNotEqual)
	}

	allPodStatus := []manager.PodStatus{}
	allPodStatus = append(allPodStatus, rtc.PdStatus...)
	allPodStatus = append(allPodStatus, rtc.TidbStatus...)
	allPodStatus = append(allPodStatus, rtc.TikvStatus...)

	for _, podStatus := range allPodStatus {
		if podStatus.Status != string(v1.PodRunning) {
			return errors.Trace(PodsNotAllRunning)
		}
	}
	if !rtc.Initialized {
		//export error info
		Errorf("TidbCluster initialization has not completed")
		return errors.Trace(InitNotCompleted)
	}
	//TODO: need to check monitor pod's status
	return cm.CheckTidbConn(tc.RootPassword)
}

func (cm *ClusterManager) RestoreData(clusterName, backupDir, fullPath string, nodeSelector map[string]string) error {
	cmds := []string{
		"/loader",
		"-h", "standalone-tidb",
		"-P", "4000",
		"-u", "root",
		"-t", "32",
		"-d", fullPath,
	}

	job := &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name: RestoreJobName,
		},
		Spec: batchv1.JobSpec{
			Template: v1.PodTemplateSpec{
				Spec: v1.PodSpec{
					RestartPolicy: v1.RestartPolicyOnFailure,
					NodeSelector:  nodeSelector,
					Containers: []v1.Container{
						{
							Name:    "loader",
							Image:   restoreImage,
							Command: cmds,
							VolumeMounts: []v1.VolumeMount{
								{Name: "data", MountPath: backupDir},
							},
						},
					},
					Volumes: []v1.Volume{
						{
							Name: "data",
							VolumeSource: v1.VolumeSource{
								HostPath: &v1.HostPathVolumeSource{Path: backupDir},
							},
						},
					},
				},
			},
		},
	}
	_, err := cm.Cli.BatchV1().Jobs(clusterName).Create(job)
	if err != nil {
		return errors.Trace(err)
	}
	return nil
}

// GetAllTiDBPods gets all TiDB pods of a TidbCluster
func (cm *ClusterManager) GetAllTiDBPods(ns, clusterName string) (*v1.PodList, error) {
	return cm.getAllPodsFor(tictl.TiDBPodType, ns, clusterName)
}

func (cm *ClusterManager) getAllPodsFor(podType tictl.PodType, ns, clusterName string) (*v1.PodList, error) {
	return cm.Cli.CoreV1().Pods(ns).List(metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			label.New().Cluster(clusterName).App(string(podType)).Labels(),
		).String(),
	})
}

// PodName returns pod name with namespace
func (cm *ClusterManager) PodName(pod *v1.Pod) string {
	return fmt.Sprintf("%s/%s", pod.GetNamespace(), pod.GetName())
}

// IsPodRunning returns whether the pod is running and is ready in PD cluster
func (cm *ClusterManager) IsPodRunning(pod *v1.Pod) (bool, error) {
	if pod.Status.Phase != v1.PodRunning {
		return false, errors.Errorf("pod: %s is not running", cm.PodName(pod))
	}

	var ownerRef *metav1.OwnerReference
	for _, or := range pod.OwnerReferences {
		if or.Kind == tsctl.ControllerKind.Kind {
			ownerRef = &or
		}
	}
	if ownerRef == nil {
		return false, errors.Errorf("pod: %s don't have OwnerReferences", cm.PodName(pod))
	}

	tidbset, err := cm.Cli.PingcapV1().TidbSets(pod.GetNamespace()).Get(ownerRef.Name, metav1.GetOptions{})
	if err != nil {
		return false, err
	}

	if strings.HasSuffix(tidbset.Name, "-pdset") {
		return tidbset.IsReady(pod), nil
	}

	if strings.HasSuffix(tidbset.Name, "-tidbset") {
		return tidbset.IsReady(pod), nil
	}

	if strings.HasSuffix(tidbset.Name, "-tikvset") {
		return tidbset.IsReady(pod), nil
	}

	return false, errors.New("this pod is not a tidb pod")
}