package task

import (
	"context"
	"flag"
	"fmt"
	"os"
	"sync"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	backoff "github.com/cenkalti/backoff/v4"
	"github.com/pingcap/errors"
	log "github.com/pingcap/log"
	"go.uber.org/multierr"
	"go.uber.org/zap"
	batchv1 "k8s.io/api/batch/v1"
	v1 "k8s.io/api/core/v1"
	k8sapierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/retry"

	"github.com/pingcap-inc/clinic/pkg/config"
	"github.com/pingcap-inc/clinic/pkg/importer/utils"
)

const (
	// k8s
	labelKeyComponent = "app.kubernetes.io/component"
	labelKeyManagedBy = "app.kubernetes.io/managed-by"
	labelKeyInstance  = "app.kubernetes.io/instance"
	labelKeyName      = "app.kubernetes.io/name"

	// import job
	labelKeyOrgID      = "clinic.pingcap.com/organization-id"
	labelKeyJobID      = "clinic.pingcap.com/job-id" // jobid == fnvhash(file uuid)
	labelKeyJobName    = "clinic.pingcap.com/job-name"
	labelKeyJobPvcName = "clinic.pingcap.com/job-pvc-name"
	labelKeyFileUUID   = "clinic.pingcap.com/file-uuid"

	importJobNamePrefix = "importer"
	importJobPvcPrefix  = "rebuild-data"
	importJobConfigMap  = "importer-config"
)

var (
	improtJoblabel = map[string]string{
		labelKeyComponent: "clinic",
		labelKeyManagedBy: "importer-manager",
		labelKeyInstance:  "import-job",
	}

	improtJobPvclabel = map[string]string{
		labelKeyComponent: "clinic",
		labelKeyManagedBy: "importer-manager",
		labelKeyInstance:  "import-job-pvc",
	}
)

// todo@toto change to yaml loading
type Job struct {
	ID             string
	FileUUID       string
	OrganizationID string
	Name           string // default: app: orgID-Importer-jobID
	NameSpace      string
	ImageName      string
	ConfigMap      string
	PvcName        string
	Param          map[string]string
}

type K8sClient interface {
	ListJob(label map[string]string) []batchv1.Job
	CleanResource()
	CreateJob(req *CreateJobRequest) error
}

var DummyK8sClient = &dummyK8sClient{}

type dummyK8sClient struct{}

func (kc *dummyK8sClient) ListJob(_ map[string]string) []batchv1.Job {
	return nil
}

func (kc *dummyK8sClient) CreateJob(_ *CreateJobRequest) error {
	return nil
}

func (kc *dummyK8sClient) CleanResource() {}

type CreateJobRequest struct {
	OrganizationID string
	PvcSize        int
	FileUUID       string
	Concurrency    int
	Param          map[string]string
}

type K8sClientV1 struct {
	cs                    kubernetes.Clientset // dynamic.Interface
	kconfig               config.K8sConfig
	importerConfigMapName string
	lock                  sync.Mutex
}

func NewK8sClient(kconfig config.K8sConfig) (*K8sClientV1, error) {
	importerConfgiMapName := os.Getenv("IMPORTER_CONFIGMAP_NAME")
	if len(importerConfgiMapName) == 0 {
		return nil, errors.New("env IMPORTER_CONFIGMAP_NAME is not set")
	}
	k8sclient := &K8sClientV1{
		lock:                  sync.Mutex{},
		kconfig:               kconfig,
		importerConfigMapName: importerConfgiMapName,
	}
	var config *rest.Config
	var err error
	if !kconfig.InCluster {
		kubeconfig := flag.String("kubeconfig", kconfig.Path, "absolute path to the kubeconfig file")
		config, err = clientcmd.BuildConfigFromFlags("", *kubeconfig)
		if err != nil {
			log.Error("init client failed")
			return nil, err
		}
	} else {
		config, err = rest.InClusterConfig()
		if err != nil {
			log.Error("init client failed")
			return nil, err
		}
	}

	// http.StatusOK
	config.Timeout = time.Second * 5
	// config.
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		log.Error("init client failed")
		return nil, err
	}

	k8sclient.cs = *clientset
	return k8sclient, nil
}

func (kc *K8sClientV1) CleanResource() {
	kc.lock.Lock()

	// delete jobs
	for _, j := range kc.ListFinishedOrUnExpectedJobs() {
		if err := kc.DeleteJobAndPvc(j); err != nil {
			log.Error("Failed to delete job or pvc.",
				zap.Error(err),
				zap.String("job id", j.Labels[labelKeyJobID]),
				zap.String("org id", j.Labels[labelKeyOrgID]),
				zap.String("file uuid", j.Labels[labelKeyFileUUID]))
		}
	}

	// after delete job , can release lock
	kc.lock.Unlock()

	if err := kc.DeletePvcs(); err != nil {
		log.Error("Failed to delete pvc.", zap.Error(err))
	}
}

func (kc *K8sClientV1) DeleteJobAndPvc(job batchv1.Job) error {
	var errs error
	if err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
		fg := metav1.DeletePropagationForeground
		return kc.cs.BatchV1().Jobs(job.Namespace).Delete(context.TODO(), job.Name, metav1.DeleteOptions{
			PropagationPolicy: &fg,
		})
	}); err != nil {
		errs = multierr.Append(errs, errors.Annotate(err, "failed to delete importer job"))
	}

	if err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
		return kc.cs.CoreV1().PersistentVolumeClaims(kc.kconfig.Namespace).Delete(context.TODO(), getPvcNameByJob(job), metav1.DeleteOptions{})
	}); err != nil {
		errs = multierr.Append(errs, errors.Annotate(err, "failed to delete importer job pvc"))
	}
	return errs
}

// DeletePvcs delete pvc after DeleteJobAndPvc failed
func (kc *K8sClientV1) DeletePvcs() error {
	labelSelector := labels.SelectorFromSet(labels.Set(improtJobPvclabel)).String()
	list, err := kc.cs.CoreV1().PersistentVolumeClaims(kc.kconfig.Namespace).List(context.TODO(), metav1.ListOptions{LabelSelector: labelSelector})
	if err != nil {
		return err
	}

	for _, pvc := range list.Items {
		if err = kc.DeletePvc(pvc); err != nil && !k8sapierrors.IsNotFound(err) {
			log.Error("delete unexpected pvc failed",
				zap.String("pvc", pvc.Name),
				zap.String("org id", pvc.Labels[labelKeyOrgID]),
				zap.String("job id", pvc.Labels[labelKeyJobID]),
				zap.String("file uuid", pvc.Labels[labelKeyFileUUID]),
				zap.Error(err))
		}
	}
	return nil
}

func (kc *K8sClientV1) DeletePvc(pvc v1.PersistentVolumeClaim) error {
	return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
		_, err := kc.cs.BatchV1().Jobs(pvc.Namespace).Get(context.TODO(), getJobNameByPvc(pvc), metav1.GetOptions{})
		// job is existed
		if err == nil {
			return nil
		}

		if !k8sapierrors.IsNotFound(err) {
			return err
		}

		// Job is not existed, delete pvc
		return kc.cs.CoreV1().PersistentVolumeClaims(kc.kconfig.Namespace).Delete(context.TODO(), pvc.Name, metav1.DeleteOptions{})
	})
}

// ListFinishedOrUnExpectedJobs list finished or unExpected jobs
func (kc *K8sClientV1) ListFinishedOrUnExpectedJobs() []batchv1.Job {
	jobs := make([]batchv1.Job, 0)

	allJobs := kc.ListJob(improtJoblabel)

	for _, j := range allJobs {
		if isJobFinished(j) || kc.isJobUnExpected(j) {
			jobs = append(jobs, j)
		}
	}

	return jobs
}

func (kc *K8sClientV1) ListAllJobs() []batchv1.Job {
	return kc.ListJob(improtJoblabel)
}

func (kc *K8sClientV1) ListJobsByOrgID(orgID string) []batchv1.Job {
	label := map[string]string{
		labelKeyOrgID: orgID,
	}

	for k, v := range improtJoblabel {
		label[k] = v
	}

	return kc.ListJob(label)
}

func (kc *K8sClientV1) ListJobsByFileUUID(fileUUID string) []batchv1.Job {
	label := map[string]string{
		labelKeyFileUUID: fileUUID,
	}

	for k, v := range improtJoblabel {
		label[k] = v
	}

	return kc.ListJob(label)
}

func (kc *K8sClientV1) ListJob(label map[string]string) []batchv1.Job {
	labelSelector := labels.SelectorFromSet(labels.Set(label)).String()

	jobs := make([]batchv1.Job, 0)
	jobsc := kc.cs.BatchV1().Jobs(kc.kconfig.Namespace)
	list, err := jobsc.List(context.TODO(), metav1.ListOptions{LabelSelector: labelSelector})
	if err != nil {
		log.Error("list jobs failed", zap.Error(err))
		return []batchv1.Job{}
	}
	jobs = append(jobs, list.Items...)
	return jobs
}

func isJobFinished(j batchv1.Job) bool {
	if j.Status.StartTime == nil {
		return false
	}

	// have pending and running pods
	if j.Status.Active != 0 {
		return false
	}

	// creation time is less than 30s
	if j.CreationTimestamp.Time.Add(30 * time.Second).After(time.Now()) {
		return false
	}

	return true
}

func (kc *K8sClientV1) isJobUnExpected(j batchv1.Job) bool {
	if j.Status.StartTime == nil {
		return false
	}

	if j.Status.StartTime.After(time.Now().Add(-10 * time.Minute)) {
		return false
	}

	// key:job-name  k8s controller auto-generation
	pods, err := kc.cs.CoreV1().Pods(j.Namespace).List(context.TODO(), metav1.ListOptions{
		LabelSelector: fmt.Sprintf("job-name=%s", j.Name),
	})
	if err != nil {
		log.Error("get pods failed", zap.String("job", j.Name), zap.Error(err))
		return false
	}

	unExpected := false
	for _, pod := range pods.Items {
		if isPodInUnexpectedStatus(pod) {
			unExpected = true
		}
	}

	return unExpected
}

// if the pod wait for scheduling, we think it is in a normal state
func isPodInUnexpectedStatus(p v1.Pod) bool {
	if p.Status.Phase != v1.PodPending {
		return false
	}

	normalStatus := false
	for _, condition := range p.Status.Conditions {
		if condition.Type == v1.PodScheduled && condition.Status == v1.ConditionFalse {
			normalStatus = true
		}
	}

	return !normalStatus
}

func (kc *K8sClientV1) CreateJob(req *CreateJobRequest) error {
	kc.lock.Lock()
	defer kc.lock.Unlock()

	// job already exists
	if len(kc.ListJobsByFileUUID(req.FileUUID)) > 0 {
		return nil
	}

	// Concurrency
	currJobs := kc.ListJobsByOrgID(req.OrganizationID)
	if len(currJobs) >= req.Concurrency {
		log.Warn("the organization has reached its maximum number of jobs",
			zap.Any("orgID", req.OrganizationID),
			zap.Any("maximum number of jobs", req.Concurrency))
		return nil
	}

	j := kc.FormatJob(req)
	jobs := kc.cs.BatchV1().Jobs(j.NameSpace)
	var backOffLimit int32 = 0

	jobSpec := &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      j.Name,
			Namespace: j.NameSpace,
			Labels:    j.getImportJobLabel(),
		},
		Spec: batchv1.JobSpec{
			// TTLSecondsAfterFinished: aws.Int32(60),
			Template: v1.PodTemplateSpec{
				Spec: v1.PodSpec{
					ServiceAccountName: "clinic",
					Containers: []v1.Container{
						{
							Name:    j.Name,
							Image:   j.ImageName,
							Command: FormatCmd(req.Param),
							TTY:     true,
							VolumeMounts: []v1.VolumeMount{
								{
									Name:      importJobConfigMap,
									MountPath: "/etc/config",
									ReadOnly:  true,
								},
								{
									Name:      "data",
									MountPath: "/data",
									ReadOnly:  false,
								},
							},
							Resources: v1.ResourceRequirements{
								Requests: map[v1.ResourceName]resource.Quantity{
									v1.ResourceMemory: resource.MustParse("8Gi"),
									v1.ResourceCPU:    *resource.NewMilliQuantity(7000, resource.DecimalSI),
								},
								Limits: map[v1.ResourceName]resource.Quantity{
									v1.ResourceMemory: resource.MustParse("15Gi"),
									v1.ResourceCPU:    *resource.NewMilliQuantity(7500, resource.DecimalSI),
								},
							},
						},
					},
					Volumes: []v1.Volume{
						{
							Name: importJobConfigMap,
							VolumeSource: v1.VolumeSource{
								ConfigMap: &v1.ConfigMapVolumeSource{
									LocalObjectReference: v1.LocalObjectReference{
										Name: kc.importerConfigMapName,
									},
								},
							},
						},
						{
							Name: "data",
							VolumeSource: v1.VolumeSource{
								PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
									ClaimName: j.PvcName,
								},
							},
						},
					},
					RestartPolicy: v1.RestartPolicyNever,
					NodeSelector: map[string]string{
						"clinic.node.type": "importer",
					},
					Tolerations: []v1.Toleration{
						{
							Key:      "clinic.service",
							Operator: "Equal",
							Value:    "importer",
							Effect:   "NoExecute",
						},
					},
				},
			},
			BackoffLimit: &backOffLimit,
		},
	}
	_, err := jobs.Create(context.TODO(), jobSpec, metav1.CreateOptions{})
	if err != nil {
		// check if the job already exists
		// if so, we will return without error
		if k8sapierrors.IsAlreadyExists(err) {
			return nil
		}
		log.Error("Failed to create K8s job.",
			zap.Error(err),
			zap.String("org id", j.OrganizationID),
			zap.String("job id", j.ID),
			zap.String("file uuid", j.FileUUID),
		)
		return err
	}

	kc.esureImporterJob(j)

	return kc.CreatePVC(j, req.PvcSize)
}

func (kc *K8sClientV1) esureImporterJob(j Job) {
	b := backoff.NewExponentialBackOff()
	b.MaxElapsedTime = 10 * time.Second
	b.InitialInterval = 50 * time.Millisecond
	b.MaxInterval = time.Second
	b.Reset()

	// esure job
	err := backoff.Retry(func() error {
		_, err := kc.cs.BatchV1().Jobs(kc.kconfig.Namespace).Get(context.TODO(), j.Name, metav1.GetOptions{})
		return err
	}, b)

	if err != nil {
		log.Warn("Failed to create K8s job.",
			zap.Error(err),
			zap.String("org id", j.OrganizationID),
			zap.String("job id", j.ID),
			zap.String("file uuid", j.FileUUID))
	} else {
		log.Info("create job successful", zap.String("job id", j.ID))
	}
}

func (kc *K8sClientV1) CreatePVC(j Job, size int) error {
	pvc := &v1.PersistentVolumeClaim{
		ObjectMeta: metav1.ObjectMeta{
			Name:      j.PvcName,
			Namespace: j.NameSpace,
			Labels:    j.getImportJobPvcLabel(),
		},
		Spec: v1.PersistentVolumeClaimSpec{
			Resources: v1.ResourceRequirements{
				Requests: map[v1.ResourceName]resource.Quantity{
					v1.ResourceStorage: resource.MustParse(fmt.Sprintf("%d", size) + "Gi"),
				},
			},
			AccessModes: []v1.PersistentVolumeAccessMode{
				v1.ReadWriteOnce,
			},
			StorageClassName: aws.String("gp2-delete"),
		},
	}
	_, err := kc.cs.CoreV1().PersistentVolumeClaims(j.NameSpace).Create(context.TODO(), pvc, metav1.CreateOptions{})
	if err != nil {
		// check if the pvc already exists
		// if so, we will return without error
		if k8sapierrors.IsAlreadyExists(err) {
			return nil
		}
		log.Error("Failed to create K8s pvc.",
			zap.String("pvc", pvc.Name),
			zap.String("org id", j.OrganizationID),
			zap.String("job id", j.ID),
			zap.String("file uuid", j.FileUUID),
			zap.Error(err))
		return err
	}

	log.Info("create pvc successful", zap.String("job id", j.ID))
	return nil
}

func (kc *K8sClientV1) FormatJob(req *CreateJobRequest) Job {
	jobID := utils.FnvHash(req.FileUUID)

	return Job{
		ID:             jobID,
		FileUUID:       req.FileUUID,
		OrganizationID: req.OrganizationID,
		Name:           getJobName(jobID),
		NameSpace:      kc.kconfig.Namespace,
		ImageName:      kc.kconfig.ImageName,
		Param:          req.Param,
		PvcName:        getPvcName(jobID),
	}
}

func (j *Job) getImportJobLabel() map[string]string {
	return j.getLabel("job")
}

func (j *Job) getImportJobPvcLabel() map[string]string {
	return j.getLabel("pvc")
}

func (j *Job) getLabel(t string) map[string]string {
	label := map[string]string{
		labelKeyOrgID:      j.OrganizationID,
		labelKeyJobID:      j.ID,
		labelKeyJobName:    j.Name,
		labelKeyJobPvcName: j.PvcName,
		labelKeyFileUUID:   j.FileUUID,
	}

	switch t {
	case "pvc":
		label[labelKeyName] = j.PvcName
		for k, v := range improtJobPvclabel {
			label[k] = v
		}
	case "job":
		label[labelKeyName] = j.Name
		for k, v := range improtJoblabel {
			label[k] = v
		}
	}
	return label
}

func getJobName(jobID string) string {
	return fmt.Sprintf("%s-%s", importJobNamePrefix, jobID)
}

func getPvcName(jobID string) string {
	return fmt.Sprintf("%s-%s", importJobPvcPrefix, jobID)
}

func getJobNameByPvc(pvc v1.PersistentVolumeClaim) string {
	return pvc.Labels[labelKeyJobName]
}

func getPvcNameByJob(job batchv1.Job) string {
	return job.Labels[labelKeyJobPvcName]
}

func FormatCmd(param map[string]string) []string {
	cmds := make([]string, 0)
	cmds = append(cmds, "./importer")
	for k, v := range param {
		cmds = append(cmds, k+"="+v)
	}
	return cmds
}