package k8s

import (
	"context"
	"strings"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/schrodinger/pkg/test_client"
	"github.com/pingcap/schrodinger/pkg/types"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap.com/v1alpha1"
	"github.com/pingcap/tidb-operator/pkg/controller"
	"github.com/pingcap/tidb-operator/pkg/label"
	"github.com/pingcap/tidb-operator/pkg/util"
	"k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/kubernetes"
)

// getClusterAccessInfo returns cluster information if cluster is retry,
// or return error if timeout
func (c *ClusterController) getClusterAccessInfo(ctx context.Context, cluster *types.Cluster, waitOK bool) error {
	if !waitOK {
		return c.genCluster(cluster)
	}

	err := wait.Poll(checkInterval, maxWaitTimeout, func() (bool, error) {
		select {
		case <-ctx.Done():
			return false, errors.Trace(ctx.Err())
		default:
			return c.isClusterRunning(cluster)
		}
	})
	if err != nil {
		log.Errorf("waiting for tidb cluster is running failed: %v", err)
		return errors.Trace(err)
	}

	return c.genCluster(cluster)
}

// getClusterStatus returns cluster status,
func (c *ClusterController) getClusterStatus(ctx context.Context, cluster *types.Cluster) types.ClusterState {
	running, err := c.isClusterRunning(cluster)
	if err != nil {
		log.Errorf("namespace: %s, check tidb cluster status failed: %v", cluster.Namespace, err)
	}

	if running {
		return types.ClusterStateRunning
	}

	return types.ClusterStateUnknown
}

func (c *ClusterController) setClusterStatus(ctx context.Context, cluster *types.Cluster) {
	cluster.State = c.getClusterStatus(ctx, cluster)
}

func (c *ClusterController) isClusterRunning(cluster *types.Cluster) (bool, error) {
	tc, err := c.ClusterCli.PingcapV1alpha1().TidbClusters(cluster.Namespace).Get(cluster.Name, metav1.GetOptions{})
	if err != nil {
		log.Errorf("namespace: %s, get tidbcluster failed: %v", cluster.Namespace, err)
		return false, nil
	}

	running := true

	if tc.Spec.PD.Replicas > 0 {
		running, err = pdMemberRunning(c.K8sCli, tc)
		if err != nil || !running {
			log.Infof("namespace: %s, pds not running", cluster.Namespace)
			return false, nil
		}
	}

	if tc.Spec.TiKV.Replicas > 0 {
		running, err = tikvMemberRunning(c.K8sCli, tc)
		if err != nil || !running {
			log.Infof("namespace: %s, tikvs not running", cluster.Namespace)
			return false, nil
		}
	}

	// TODO: Start pump/drainer by tidb-sch-ctl. not tidb-sch-ctl not support start pump/drainer
	// when enable binlog but pump/drainer not started. tidb's port can not be connected
	// this is a temp code
	if strings.Contains(cluster.Namespace, "binlog") {
		return true, nil
	}

	if tc.Spec.TiDB.Replicas > 0 {
		running, err = tidbMemberRunning(c.K8sCli, tc)
		if err != nil || !running {
			log.Infof("namespace: %s, tidbs not running", cluster.Namespace)
			return false, nil
		}
	}

	if tc.Spec.PD.Replicas+tc.Spec.TiKV.Replicas > 0 {
		var synced bool
		synced, err = reclaimPolicySynced(c.K8sCli, tc)
		if err != nil || !synced {
			log.Infof("namespace: %s, claim not synced", cluster.Namespace)
			return false, nil
		}
	}

	if err = c.handleConfigSet(cluster); err != nil {
		log.Errorf("handle config set for cluster %s failed %v", cluster.Namespace, err)
		return false, nil
	}

	return true, nil
}

func (c *ClusterController) genCluster(cluster *types.Cluster) error {
	log.Infof("genrate cluster namespace %s, name %s", cluster.Namespace, cluster.Name)
	labs := label.New().Instance(cluster.Namespace).Labels()
	option := metav1.ListOptions{LabelSelector: labels.SelectorFromSet(labs).String()}
	pods, err := c.K8sCli.CoreV1().Pods(cluster.Namespace).List(option)
	if err != nil {
		return errors.Trace(err)
	}

	cluster.PdStatus = cluster.PdStatus[:0]
	cluster.TikvStatus = cluster.TikvStatus[:0]
	cluster.TidbStatus = cluster.TidbStatus[:0]

	for _, pod := range pods.Items {
		if pod.Status.Phase == v1.PodFailed {
			log.Errorf("namespace %s pod name %s phase %v reason %v", cluster.Namespace, pod.ObjectMeta.GetName(), pod.Status.Phase, pod.Status.Reason)
			continue
		}
		if pod.ObjectMeta.Labels != nil {
			switch pod.ObjectMeta.Labels[label.ComponentLabelKey] {
			case label.PDLabelVal:
				cluster.PdStatus = append(cluster.PdStatus, genPodStatus(pod, client.PD, cluster.Status))
			case label.TiKVLabelVal:
				cluster.TikvStatus = append(cluster.TikvStatus, genPodStatus(pod, client.TiKV, cluster.Status))
			case label.TiDBLabelVal:
				cluster.TidbStatus = append(cluster.TidbStatus, genPodStatus(pod, client.TiDB, cluster.Status))
			case "tidb-monitor", "configure-grafana", "privileged-tidb":
				log.Debugf("tidb monitor: %s, namespace: %s", pod.ObjectMeta.Labels[label.ComponentLabelKey], cluster.Namespace)
			default:
				log.Warnf("unexpected pod type: %s, namespace: %s", pod.ObjectMeta.Labels[label.ComponentLabelKey], cluster.Namespace)
			}
		}
	}
	serviceNames := map[string]string{
		"grafana":    getName(cluster.Name, "grafana"),
		"prometheus": getName(cluster.Name, "prometheus"),
		"tidb":       getName(cluster.Name, "tidb"),
		"pd":         getName(cluster.Name, "pd"),
	}
	for kind, name := range serviceNames {
		svc, err := c.K8sCli.CoreV1().Services(cluster.Namespace).Get(name, metav1.GetOptions{})
		if err != nil {
			log.Warnf("failed to get service: %s, namespace: %s, error: %v", name, cluster.Namespace, err)
			continue
		}
		service := c.getService(svc)
		switch kind {
		case "grafana":
			cluster.GrafanaService = service
		case "prometheus":
			cluster.PrometheusService = service
		case "tidb":
			cluster.TidbService = service
		case "pd":
			cluster.PDService = service
		default:
			log.Warnf("unexpected service kind: %s, namespace: %s", kind, cluster.Namespace)
		}
	}

	return nil
}

func (c *ClusterController) getService(svc *v1.Service) types.Service {
	service := types.Service{
		ClusterIP:   svc.Spec.ClusterIP,
		ClusterPort: int(svc.Spec.Ports[0].Port),
	}
	if svc.Spec.Type == v1.ServiceTypeNodePort || svc.Spec.Type == v1.ServiceTypeLoadBalancer {
		service.NodePort = int(svc.Spec.Ports[0].NodePort)
		service.NodeIP = c.getLatestNodes()
	}
	if svc.Spec.Type == v1.ServiceTypeLoadBalancer && len(svc.Status.LoadBalancer.Ingress) > 0 {
		service.ExternalIP = svc.Status.LoadBalancer.Ingress[0].IP
	} else if svc.Spec.Type == v1.ServiceTypeLoadBalancer {
		log.Warnf("have not allocated ingress ip for loadBalancer, namespace: %s", svc.ObjectMeta.Namespace)
	}
	return service
}

// NodesInfo defines a set of node of k8s
type NodesInfo struct {
	Nodes     []string
	UpdatedAt time.Time
}

var nodesInfo NodesInfo

func (c *ClusterController) getLatestNodes() []string {
	if int(time.Now().Sub(nodesInfo.UpdatedAt).Seconds()) < 30 {
		return nodesInfo.Nodes
	}
	nodes, err := c.K8sCli.CoreV1().Nodes().List(metav1.ListOptions{})
	if err != nil {
		log.Errorf("failed to retrieve nodes: %v, error: %v", nodes, err)
		return nodesInfo.Nodes
	}
	nodesInfo.Nodes = []string{}
	for _, node := range nodes.Items {
		if node.Spec.Unschedulable {
			continue
		}
		for _, a := range node.Status.Addresses {
			if a.Type == v1.NodeInternalIP {
				nodesInfo.Nodes = append(nodesInfo.Nodes, a.Address)
			}
		}
	}
	nodesInfo.UpdatedAt = time.Now()
	return nodesInfo.Nodes
}

func genPodStatus(pod v1.Pod, typeName string, clusterStatus v1alpha1.TidbClusterStatus) types.PodStatus {
	status := types.PodStatus{
		Name:   pod.ObjectMeta.GetName(),
		PodIP:  pod.Status.PodIP,
		NodeIP: pod.Status.HostIP,
		Status: string(pod.Status.Phase),
	}

	for _, container := range pod.Status.ContainerStatuses {
		if container.Name == typeName {
			status.Image = container.Image
			status.ContainerID = strings.TrimPrefix(container.ContainerID, "docker://")
			break
		}
	}

	status.ProcessStatus = util.StoreDownState

	switch typeName {
	case client.PD:
		for _, pd := range clusterStatus.PD.Members {
			if pd.Name == status.Name && pd.Health {
				status.ProcessStatus = util.StoreUpState
			}
		}
	case client.TiDB:
		for _, tidb := range clusterStatus.TiDB.Members {
			if tidb.Name == status.Name && tidb.Health {
				status.ProcessStatus = util.StoreUpState
			}
		}
	case client.TiKV:
		for _, tikv := range clusterStatus.TiKV.Stores {
			if tikv.PodName == status.Name {
				status.ProcessStatus = tikv.State
			}
		}
	}

	return status
}

func pdMemberRunning(kubeCli kubernetes.Interface, tc *v1alpha1.TidbCluster) (bool, error) {
	ns := tc.GetNamespace()
	clusterName := tc.Name
	pdSetName := controller.PDMemberName(clusterName)
	pdSet, err := kubeCli.AppsV1().StatefulSets(ns).Get(pdSetName, metav1.GetOptions{})
	if err != nil {
		log.Errorf("namespace: %s, get pd statefulSet failed: %v", ns, err)
		return false, nil
	}

	log.Infof("namespace: %s, pdSet.Status: %+v", ns, pdSet.Status)

	if tc.Status.PD.StatefulSet == nil {
		log.Infof("namespace: %s, tc.Status.PD.StatefulSet is nil", ns)
		return false, nil
	}

	if *pdSet.Spec.Replicas != tc.Spec.PD.Replicas {
		log.Infof("namespace: %s, pdSet.Spec.Replicas(%d) != tc.Spec.PD.Replicas(%d)",
			ns, *pdSet.Spec.Replicas, tc.Spec.PD.Replicas)
		return false, nil
	}

	if pdSet.Status.ReadyReplicas != tc.Spec.PD.Replicas {
		log.Infof("namespace: %s, pdSet.Status.ReadyReplicas(%d) != %d",
			ns, pdSet.Status.ReadyReplicas, tc.Spec.PD.Replicas)
		return false, nil
	}

	if len(tc.Status.PD.Members) != int(tc.Spec.PD.Replicas) {
		log.Infof("namespace: %s, tc.Status.PD.Members count(%d) != %d",
			ns, len(tc.Status.PD.Members), tc.Spec.PD.Replicas)
		return false, nil
	}

	if pdSet.Status.ReadyReplicas != pdSet.Status.Replicas {
		log.Infof("namespace: %s, pdSet.Status.ReadyReplicas(%d) != pdSet.Status.Replicas(%d)",
			ns, pdSet.Status.ReadyReplicas, pdSet.Status.Replicas)
		return false, nil
	}

	for _, member := range tc.Status.PD.Members {
		if !member.Health {
			log.Infof("namespace: %s, pd member(%s) is not health", ns, member.ID)
			return false, nil
		}
	}

	_, err = kubeCli.CoreV1().Services(ns).Get(controller.PDMemberName(clusterName), metav1.GetOptions{})
	if err != nil {
		log.Infof("namespace: %s, get pd service by pd memeber name failed: %v", ns, err)
		return false, nil
	}
	_, err = kubeCli.CoreV1().Services(ns).Get(controller.PDPeerMemberName(clusterName), metav1.GetOptions{})
	if err != nil {
		log.Infof("namespace: %s, get pd service by pd peer memeber by name failed: %v", ns, err)
		return false, nil
	}

	return true, nil
}

func tikvMemberRunning(kubeCli kubernetes.Interface, tc *v1alpha1.TidbCluster) (bool, error) {
	ns := tc.GetNamespace()
	clusterName := tc.Name
	tikvSetName := controller.TiKVMemberName(clusterName)

	tikvSet, err := kubeCli.AppsV1().StatefulSets(ns).Get(tikvSetName, metav1.GetOptions{})
	if err != nil {
		log.Errorf("namespace: %s, get tikv statefulset failed: %v", ns, err)
		return false, nil
	}

	log.Infof("namespace: %s, tikvSet.Status: %+v", ns, tikvSet.Status)

	if tc.Status.TiKV.StatefulSet == nil {
		log.Infof("namespace: %s, tc.Status.TiKV.StatefulSet is nil", ns)
		return false, nil
	}

	if *tikvSet.Spec.Replicas != tc.Spec.TiKV.Replicas {
		log.Infof("namespace: %s, tikvSet.Spec.Replicas(%d) != tc.Spec.TiKV.Replicas(%d)",
			ns, *tikvSet.Spec.Replicas, tc.Spec.TiKV.Replicas)
		return false, nil
	}

	if tikvSet.Status.ReadyReplicas != tc.Spec.TiKV.Replicas {
		log.Infof("namespace: %s, tikvSet.Status.ReadyReplicas(%d) != %d",
			ns, tikvSet.Status.ReadyReplicas, tc.Spec.TiKV.Replicas)
		return false, nil
	}

	if len(tc.Status.TiKV.Stores) != int(tc.Spec.TiKV.Replicas) {
		log.Infof("namespace: %s, tc.Status.TiKV.Stores.count(%d) != %d",
			ns, len(tc.Status.TiKV.Stores), tc.Spec.TiKV.Replicas)
		return false, nil
	}

	if tikvSet.Status.ReadyReplicas != tikvSet.Status.Replicas {
		log.Infof("namespace: %s, tikvSet.Status.ReadyReplicas(%d) != tikvSet.Status.Replicas(%d)",
			ns, tikvSet.Status.ReadyReplicas, tikvSet.Status.Replicas)
		return false, nil
	}

	for _, store := range tc.Status.TiKV.Stores {
		if store.State != util.StoreUpState {
			log.Infof("namespace: %s, store(%s) state != %s", ns, store.ID, util.StoreUpState)
			return false, nil
		}
	}

	_, err = kubeCli.CoreV1().Services(ns).Get(controller.TiKVPeerMemberName(clusterName), metav1.GetOptions{})
	if err != nil {
		log.Infof("namespace: %s, get tikv service by tikv perr member name failed: %v", ns, err)
		return false, nil
	}

	return true, nil
}

func tidbMemberRunning(kubeCli kubernetes.Interface, tc *v1alpha1.TidbCluster) (bool, error) {
	ns := tc.GetNamespace()
	clusterName := tc.Name
	tidbSetName := controller.TiDBMemberName(clusterName)

	tidbSet, err := kubeCli.AppsV1().StatefulSets(ns).Get(tidbSetName, metav1.GetOptions{})
	if err != nil {
		log.Errorf("namespace: %s, get tidb statefulset failed: %v", ns, err)
		return false, nil
	}

	log.Infof("namespace: %s, tidbSet.Status: %+v", ns, tidbSet.Status)

	if tc.Status.TiDB.StatefulSet == nil {
		log.Infof("namespace: %s, tc.Status.TiDB.StatefulSet is nil", ns)
		return false, nil
	}

	if *tidbSet.Spec.Replicas != tc.Spec.TiDB.Replicas {
		log.Infof("namespace: %s, tidbSet.Spec.Replicas(%d) != tc.Spec.TiDB.Replicas(%d)",
			ns, *tidbSet.Spec.Replicas, tc.Spec.TiDB.Replicas)
		return false, nil
	}

	if tidbSet.Status.ReadyReplicas != tc.Spec.TiDB.Replicas {
		log.Infof("namespace: %s, tidbSet.Status.ReadyReplicas(%d) != %d",
			ns, tidbSet.Status.ReadyReplicas, tc.Spec.TiDB.Replicas)
		return false, nil
	}

	if tidbSet.Status.ReadyReplicas != tidbSet.Status.Replicas {
		log.Infof("namespace: %s, tidbSet.Status.ReadyReplicas(%d) != tidbSet.Status.Replicas(%d)",
			ns, tidbSet.Status.ReadyReplicas, tidbSet.Status.Replicas)
		return false, nil
	}

	for _, member := range tc.Status.TiDB.Members {
		if !member.Health {
			log.Infof("namespace: %s, tidb member(%s) is not health", ns, member.Name)
			return false, nil
		}
	}

	_, err = kubeCli.CoreV1().Services(ns).Get(controller.TiDBMemberName(clusterName), metav1.GetOptions{})
	if err != nil {
		log.Infof("namespace: %s, get tidb service by tidb memeber name failed: %v", ns, err)
		return false, nil
	}

	return true, nil
}

func reclaimPolicySynced(kubeCli kubernetes.Interface, tc *v1alpha1.TidbCluster) (bool, error) {
	ns := tc.GetNamespace()
	labelSelector := label.New().Instance(ns)
	pvcList, err := kubeCli.CoreV1().PersistentVolumeClaims(ns).List(
		metav1.ListOptions{
			LabelSelector: labels.SelectorFromSet(
				labelSelector.Labels(),
			).String(),
		},
	)
	if err != nil {
		log.Errorf("namespace: %s, get pvc list failed: %v", ns, err)
		return false, nil
	}

	for _, pvc := range pvcList.Items {
		pv, err := kubeCli.CoreV1().PersistentVolumes().Get(pvc.Spec.VolumeName, metav1.GetOptions{})
		if err != nil {
			log.Errorf("namespace: %s, get pv faild: %v", ns, err)
			return false, nil
		}

		log.Infof("namespace: %v, pv: %s's persistentVolumeReclaimPolicy is %s",
			ns, pv.GetName(), pv.Spec.PersistentVolumeReclaimPolicy)
		if pv.Spec.PersistentVolumeReclaimPolicy != tc.Spec.PVReclaimPolicy {
			return false, nil
		}
	}

	return true, nil
}