package k8s

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/hashicorp/go-multierror"
	"go.uber.org/zap"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/client-go/util/retry"

	"github.com/pingcap/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/util/wait"

	apiresource "github.com/pingcap/test-infra/common/model/resource"

	"github.com/pingcap/test-infra/sdk/resource"
)

var (
	_ resource.DMCluster = &DMCluster{}

	dmGVR = schema.GroupVersionResource{Group: "pingcap.com", Version: "v1alpha1", Resource: "dmclusters"}
)

type DMCluster struct {
	KubeClientSet
	apiresource.DMCluster
	// k8s CR object
	dmcr *unstructured.Unstructured // dm-cluster CR object
}

func (d *DMCluster) Handle() apiresource.ResourceHandle {
	return d.DMCluster.ResourceHandle
}

func (d *DMCluster) Name() string { return d.dmcr.GetName() }

func (d *DMCluster) Namespace() string { return d.dmcr.GetNamespace() }

func (d *DMCluster) Exec(podName, containerName string, cmd ...string) (stdout string, stderr string, exitCode int, err error) {
	sugar.Info("exec on dm", zap.String("podName", podName), zap.String("containerName", containerName), zap.Strings("cmd", cmd))
	stdout, stderr, ec, err := d.ExecCommandInContainerWithFullOutput(podName, containerName, cmd...)
	if err != nil {
		sugar.Error("Exce error", zap.Error(err))
		err = errors.Trace(err)
	}
	sugar.Info("exec on dm result", zap.String("podName", podName), zap.String("containerName", containerName),
		zap.Strings("cmd", cmd), zap.String("stdout", stdout), zap.String("stderr", stderr))
	if ec != nil {
		exitCode = *ec
	} else {
		sugar.Warnw("cannot get the exit code, return -1")
		exitCode = -1
	}
	return
}

func (d *DMCluster) WaitForReady(timeout time.Duration) error {
	interval := 10 * time.Second
	err := wait.PollImmediate(interval, timeout, func() (done bool, err error) {
		if err := d.Sync(); err != nil {
			return false, err
		}
		return d.Ready(), nil
	})
	return errors.Trace(err)
}

func (d *DMCluster) Ready() bool {
	if len(d.Status.Conditions) == 0 {
		return false
	}
	condition := d.Status.Conditions[len(d.Status.Conditions)-1]
	return condition.Type == apiresource.DMClusterReady && condition.Reason == "Ready"
}

func (d *DMCluster) Sync() error {
	var errs error

	dmcr, err := d.DynamicClient.Resource(dmGVR).Namespace(d.KubeClientSet.Namespace).Get(context.TODO(), d.ResourceHandle.Name, metav1.GetOptions{})
	if err != nil {
		errs = multierror.Append(errs, err)
		dmcr = nil
	}
	pvcs, err := d.ClientSet.CoreV1().PersistentVolumeClaims(d.KubeClientSet.Namespace).List(context.TODO(), metav1.ListOptions{
		LabelSelector: labels.Set{"app.kubernetes.io/instance": d.ResourceHandle.Name}.AsSelector().String(),
	})
	if err != nil {
		return multierror.Append(errs, err)
	}
	// if all k8s resource created by dmcluster are not found, return an ErrorResourceNotFound error
	if len(pvcs.Items) == 0 && dmcr == nil {
		return resource.ErrorResourceNotFound{ResourceHandle: d.Handle()}
	}
	if errs != nil {
		return errs
	}

	d.dmcr = dmcr
	buf := new(bytes.Buffer)
	err = json.NewEncoder(buf).Encode(dmcr)
	if err != nil {
		return errors.Trace(err)
	}
	err = json.NewDecoder(buf).Decode(&d.DMCluster)
	if err != nil {
		return errors.Trace(err)
	}
	d.ResourceHandle.Ready = &[]bool{d.Ready()}[0]
	return nil
}

func (d *DMCluster) Upgrade(opts resource.DMClusterUpgradeOptions) error {
	now := time.Now()
	if err := d.upgradeCluster(opts); err != nil {
		return errors.Trace(err)
	}
	if opts.Timeout == 0 {
		return errors.Trace(d.WaitForReady(opts.Timeout))
	}
	left := time.Until(now.Add(opts.Timeout))
	if left <= 0 {
		return ErrUpgradeTimeout
	}
	return errors.Trace(d.WaitForReady(left))
}

func (d *DMCluster) upgradeCluster(opts resource.DMClusterUpgradeOptions) error {
	var err error
	if err = d.Sync(); err != nil {
		return err
	}
	err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
		err = unstructured.SetNestedField(d.dmcr.Object, opts.Version, "spec", "version")
		_, updateErr := d.KubeClientSet.DynamicClient.Resource(dmGVR).Namespace(d.Namespace()).Update(context.TODO(), d.dmcr, metav1.UpdateOptions{})
		if updateErr == nil {
			return nil
		}
		if err = d.Sync(); err != nil {
			return err
		}
		return updateErr
	})
	if err != nil {
		return err
	}
	now := time.Now()
	for _, component := range []resource.DMCompType{resource.DMMaster, resource.DMWorker} {
	waitUpdate:
		for {
			if err := d.Sync(); err != nil {
				return err
			}
			var baseImage, componentVersion string
			switch component {
			case resource.DMMaster:
				baseImage = d.Spec.Master.BaseImage
				componentVersion = d.Spec.Master.Version
			case resource.DMWorker:
				baseImage = d.Spec.Worker.BaseImage
				componentVersion = d.Spec.Worker.Version
			}
			if baseImage == "" {
				// although tide-operator sets the baseImage to defaultImage like pingcap/dm when spec.version isn't empty(on crd's the default webhook),
				// we'd better check it again here
				return fmt.Errorf("component %s's baseImage isn't set, please check your testbed", component.String())
			}
			if componentVersion != "" && componentVersion != opts.Version {
				// we should make clear what behaviour is user expected when spec.<master/worker>.version is set
				// and before that just fail fast here
				return fmt.Errorf("undefined behaviour, component %s's version has been set and isn't equal to the target version, please check your testbed and testcase", component.String())
			}
			upgradeImage := fmt.Sprintf("%s:%s", baseImage, opts.Version)
			switch component {
			case resource.DMMaster:
				if d.Status.Master.Image == upgradeImage && string(d.Status.Master.Phase) == normalPhase {
					break waitUpdate
				}
			case resource.DMWorker:
				if d.Status.Worker.Image == upgradeImage && string(d.Status.Worker.Phase) == normalPhase {
					break waitUpdate
				}
			}
			if opts.Timeout != 0 && time.Now().After(now.Add(opts.Timeout)) {
				return ErrUpgradeTimeout
			}
			time.Sleep(5 * time.Second)
		}
	}
	return nil
}

func (d *DMCluster) ChaosClient() resource.ChaosClient {
	return &ChaosClient{d.KubeClientSet}
}

func (d *DMCluster) ChaosSelect(selector resource.DMClusterSelectorOpts) ([]resource.ChaosSelector, error) {
	var selectors []resource.ChaosSelector
	if selector.DMClusterWorkerSelectorSpec != nil {
		spec := selector.DMClusterWorkerSelectorSpec
		switch {
		case spec.Peer != nil:
			pod, err := d.KubeClientSet.ClientSet.CoreV1().Pods(d.Namespace()).Get(context.TODO(), fmt.Sprintf("%s-dm-worker-%d", d.Name(), *spec.Peer), metav1.GetOptions{})
			if err != nil {
				return nil, err
			}
			selectors = append(selectors, &PodSelector{pod})
		case spec.All:
			pods, err := d.KubeClientSet.ClientSet.CoreV1().Pods(d.Namespace()).List(context.TODO(), metav1.ListOptions{
				LabelSelector: labels.Set{"app.kubernetes.io/instance": d.Name(), "app.kubernetes.io/component": "dm-worker"}.AsSelector().String(),
			})
			if err != nil {
				return nil, err
			}
			for _, worker := range pods.Items {
				pod := worker
				selectors = append(selectors, &PodSelector{&pod})
			}
		}
	}

	if selector.DMClusterMasterSelectorSpec != nil {
		spec := selector.DMClusterMasterSelectorSpec
		switch {
		case spec.Peer != nil:
			pod, err := d.KubeClientSet.ClientSet.CoreV1().Pods(d.Namespace()).Get(context.TODO(), fmt.Sprintf("%s-dm-master-%d", d.Name(), *spec.Peer), metav1.GetOptions{})
			if err != nil {
				return nil, err
			}
			selectors = append(selectors, &PodSelector{pod})
		case spec.All:
			pods, err := d.KubeClientSet.ClientSet.CoreV1().Pods(d.Namespace()).List(context.TODO(), metav1.ListOptions{
				LabelSelector: labels.Set{"app.kubernetes.io/instance": d.Name(), "app.kubernetes.io/component": "dm-master"}.AsSelector().String(),
			})
			if err != nil {
				return nil, err
			}
			for _, master := range pods.Items {
				pod := master
				selectors = append(selectors, &PodSelector{&pod})
			}
		}
	}
	return selectors, nil
}