package proc

import (
	"fmt"
	"strings"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/schrodinger/manager/operator/proc/k8s"
	"github.com/pingcap/schrodinger/util"
	"golang.org/x/net/context"
	"k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
)

const (
	postRetry = 2
)

// ProcsController defines for managing k8s
type ProcsController struct {
	K8sCli kubernetes.Interface
}

// NewProcsController returns a k8sproc.controller
func NewProcsController(k8scli kubernetes.Interface) *ProcsController {
	return &ProcsController{
		K8sCli: k8scli,
	}
}

// GetProcByName returns Proc by specified name
func (c *ProcsController) GetProcByName(namespace string, name string) (*k8s.Proc, error) {
	p := &k8s.Proc{
		Namespace:   namespace,
		Name:        name,
		ProcessName: name,
	}
	var err error
	p.Pod, err = util.GetPod(p.Namespace, util.FormatSnakeCase(p.Name), c.K8sCli)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if p.Pod == nil {
		return nil, errors.NotFoundf("process %s", p.Name)
	}

	log.Debugf("p %+v", p)
	p.SetAgentClient()
	p.GetState()
	return p, nil
}

//GetServiceAddr returns service by specified name
func (c *ProcsController) GetServiceAddr(namespace string, name string) (string, error) {
	addr, err := util.GetServiceAddr(namespace, name, c.K8sCli)
	return addr, errors.Trace(err)
}

//DeleteNamespace deletes namespace
func (c *ProcsController) DeleteNamespace(namespace string) error {
	return errors.Trace(util.DeleteK8sNamespace(namespace, c.K8sCli))
}

func (c *ProcsController) existsPodsInNamespace(namespace string) (bool, error) {
	options := metav1.ListOptions{}
	pods, err := util.ListPods(namespace, c.K8sCli, options)
	if err != nil {
		return false, errors.Trace(err)
	}
	return len(pods.Items) > 0, nil
}

//ListProcs list Proc under namespace
func (c *ProcsController) ListProcs(namespace string) ([]*k8s.Proc, error) {
	var ps []*k8s.Proc
	options := metav1.ListOptions{}
	pods, err := util.ListPods(namespace, c.K8sCli, options)
	if err != nil {
		return ps, errors.Trace(err)
	}

	for _, po := range pods.Items {
		pod := po
		p := &k8s.Proc{
			Name:      pod.Name,
			Namespace: pod.Namespace,
			Pod:       &pod,
		}
		p.SetAgentClient()
		p.GetState()
		ps = append(ps, p)
	}
	return ps, nil
}

func (c *ProcsController) checkOrCreateNamespace(namespace string) error {
	isExist, err := util.CheckNamespaceIsExist(namespace, c.K8sCli)
	if err != nil {
		return errors.Trace(err)
	}
	if !isExist {
		err = util.CreateK8sNamespace(namespace, c.K8sCli)
		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

// StartProc starts Proc
func (c *ProcsController) StartProc(ctx context.Context, p *k8s.Proc) error {
	err := c.checkOrCreateNamespace(p.Namespace)
	if err != nil {
		return errors.Trace(err)
	}

	if err = c.createPod(ctx, p); err != nil {
		return errors.Trace(err)
	}

	p.Pod, err = util.GetPod(p.Namespace, util.FormatSnakeCase(p.Name), c.K8sCli)
	if err != nil {
		return errors.Trace(err)
	}

	p.SetAgentClient()

	err = util.RetryOnError(ctx, postRetry, func() error {
		err = p.StartProcess()
		if err != nil && !strings.Contains(err.Error(), "already exists") {
			return errors.Trace(err)
		}
		return nil
	})
	if err != nil {
		return errors.Trace(err)
	}
	p.GetState()

	return nil
}

//DestroyProc deletes pod
func (c *ProcsController) DestroyProc(p *k8s.Proc) error {
	err := util.DeletePod(p.Namespace, p.Name, c.K8sCli)
	if err != nil {
		if !strings.Contains(err.Error(), "not found") {
			return errors.Trace(err)
		}
	}

	p.UpdateState(k8s.ProcDestroyed)

	isExist, err := c.existsPodsInNamespace(p.Namespace)
	if err != nil {
		return errors.Trace(err)
	}

	if !isExist {
		return errors.Trace(util.DeleteK8sNamespace(p.Namespace, c.K8sCli))
	}
	return nil
}

//StopProc stops Proc
func (c *ProcsController) StopProc(ctx context.Context, p *k8s.Proc) error {
	if p.Pod == nil {
		return errors.Errorf("k8sproc.%s pod is nil", p.Name)
	}

	p.SetAgentClient()
	p.GetState()

	if !p.CanStop() {
		return nil
	}

	err := util.RetryOnError(ctx, postRetry, func() error {
		err := p.StopProcess()
		if err != nil {
			return errors.Trace(err)
		}
		return nil
	})
	if err != nil {
		return errors.Trace(err)
	}
	p.UpdateState(k8s.ProcStopped)
	return nil
}

//RestartProc restarts Proc
func (c *ProcsController) RestartProc(ctx context.Context, p *k8s.Proc) error {
	_, err := util.GetPod(p.Namespace, util.FormatSnakeCase(p.Name), c.K8sCli)
	if err != nil && !strings.Contains(err.Error(), "not found") {
		return errors.Trace(err)
	}
	if err == nil {
		p.SetAgentClient()
		err := util.RetryOnError(ctx, postRetry, func() error {
			if err := p.RestartProcess(); err != nil {
				return errors.Trace(err)
			}
			return nil
		})
		p.GetState()
		return errors.Trace(err)
	}
	return c.StartProc(ctx, p)
}

//GetProcLog gets pods last lines line of log
func (c *ProcsController) GetProcLog(p *k8s.Proc, lines int64) (string, error) {
	lastLog, err := util.GetPodLog(p.Namespace, c.K8sCli, p.Pod.GetName(), &v1.PodLogOptions{TailLines: &lines})
	if err != nil {
		return "", errors.Trace(err)
	}
	fatalLog := util.FatalLogAndBelow(string(lastLog))
	if fatalLog == "" {
		return string(lastLog), nil
	}
	return fatalLog, nil
}

// Status returns k8sproc.status
func (c *ProcsController) Status(p *k8s.Proc) k8s.PState {
	var err error
	p.Pod, err = util.GetPod(p.Namespace, util.FormatSnakeCase(p.Name), c.K8sCli)
	if err != nil {
		return k8s.ProcTimeout
	}
	if p.Pod == nil {
		return k8s.ProcTimeout
	}
	p.SetAgentClient()
	return p.GetState()
}

// CheckProcIsStarted return true if it has been started
// If namespace of the k8sproc.is not exist or the pod is not exist, the k8s can be considered to have been not started.
// then gets the state of the k8sproc. if error returned contains the 'not found',
// the k8sproc.can be thought of as having been not started, otherwise it is considered to have been started.
func (c *ProcsController) CheckProcIsStarted(p *k8s.Proc) (bool, error) {
	nsIsExist, err := util.CheckNamespaceIsExist(p.Namespace, c.K8sCli)
	if err != nil {
		return false, errors.Trace(err)
	}

	if !nsIsExist {
		return false, nil
	}

	p.Pod, err = util.GetPod(p.Namespace, util.FormatSnakeCase(p.Name), c.K8sCli)
	if err != nil {
		if strings.Contains(err.Error(), "not found") {
			return false, nil
		}

		return false, errors.Trace(err)
	}

	p.SetAgentClient()

	notFound, err := p.IsNotFound()
	if err != nil {
		return false, errors.Trace(err)
	}

	return !notFound, nil
}

func (c *ProcsController) createPod(ctx context.Context, p *k8s.Proc) error {
	cmds := []string{
		"/agent",
		fmt.Sprintf("--addr=0.0.0.0:%d", p.Port),
	}
	containers := []v1.Container{
		{
			Name:    util.FormatSnakeCase(p.Name),
			Image:   p.Image,
			Command: cmds,
			Ports: []v1.ContainerPort{
				{
					ContainerPort: p.Port,
				},
			},
			Env:          p.Env,
			Resources:    p.Resources,
			VolumeMounts: p.VolumeMounts,
		},
	}

	meta := metav1.ObjectMeta{
		Name: util.FormatSnakeCase(p.Name),
	}

	_, err := util.CreatePod(p.Namespace, c.K8sCli, &v1.Pod{
		ObjectMeta: meta,
		Spec: v1.PodSpec{
			HostNetwork:  p.HostNetwork,
			Containers:   containers,
			Volumes:      p.Volumes,
			NodeSelector: util.CleanIDCLabels(p.Labels),
		},
	})
	if err != nil && !strings.Contains(err.Error(), "already exists") {
		return errors.Trace(err)
	}
	if ok := util.CheckPodIsRunning(ctx, p.Namespace, p.Name, c.K8sCli); !ok {
		return errors.Errorf("Create pod %s failed", p.Name)
	}
	return nil
}