package e2e

import (
	"fmt"
	"path/filepath"
	"time"

	"os"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap.com/v1"
	"github.com/pingcap/tidb-operator/test/e2e/framework"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/wait"
)

var backupDir = "/var/e2e-backup/"

func testBackup(f *framework.Framework) {
	clusterName := "e2e-backup"
	var pdCount int32 = 3
	var tidbCount int32 = 1
	var tikvCount int32 = 3

	backupImage, found := os.LookupEnv("BACKUP_IMAGE")
	if !found {
		backupImage = "localhost:5000/pingcap/tidb-backup:latest"
	}

	done := deployCluster(f, e2eNamespace, clusterName, pdCount, tidbCount, tikvCount)

	By(fmt.Sprintf("When full backup 's backup job is created"))
	nodes, err := f.KubeCli.CoreV1().Nodes().List(metav1.ListOptions{})
	Expect(err).NotTo(HaveOccurred())
	Expect(len(nodes.Items)).ToNot(BeZero())
	backupNode := nodes.Items[0]
	fullBakcupName := fmt.Sprintf("%s-full-backup", clusterName)
	fullbackup := &v1.TidbClusterBackup{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: e2eNamespace,
			Name:      fullBakcupName,
		},
		Spec: v1.TidbClusterBackupSpec{
			Cluster:     clusterName,
			BackupImage: backupImage,
			FullBackup: &v1.FullBackupSpec{
				BackupDir:    backupDir,
				BackupNode:   backupNode.Name,
				BackupCron:   "*/1 * * * *",
				BackupParams: map[string]string{"-t": "4"},
			},
		},
	}
	fullbackup, err = f.Cli.PingcapV1().TidbClusterBackups(e2eNamespace).Create(fullbackup)
	Expect(err).NotTo(HaveOccurred())

	By(fmt.Sprintf("Then full backup 's backup job have completed successfully "))
	err = wait.Poll(5*time.Second, 10*time.Minute, func() (bool, error) {
		fullbackup, err = f.Cli.PingcapV1().TidbClusterBackups(e2eNamespace).Get(fullBakcupName, metav1.GetOptions{})
		if err != nil {
			return false, nil
		}

		if len(fullbackup.Status.FullBackupStatus.CronJobName) == 0 {
			return false, nil
		}

		if len(fullbackup.Status.FullBackupStatus.BackupTaskStatuses) == 0 {
			return false, nil
		}

		for _, backupTaskStatus := range fullbackup.Status.FullBackupStatus.BackupTaskStatuses {
			if backupTaskStatus.State == v1.BackupFailed {
				err = fmt.Errorf("the backup status is failed,%v", backupTaskStatus)
				return true, err
			}
			if backupTaskStatus.State != v1.BackupSuccess {
				return false, nil
			}
		}
		return true, nil
	})
	Expect(err).NotTo(HaveOccurred())

	checkpods := []corev1.Pod{}
	By(fmt.Sprintf("Then we will check the backup data "))
	fullbackup, err = f.Cli.PingcapV1().TidbClusterBackups(e2eNamespace).Get(fullBakcupName, metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())
	backupTaskStatus := fullbackup.Status.FullBackupStatus.BackupTaskStatuses[0]
	checkpod := getCheckPod(backupTaskStatus.BackupNode, backupTaskStatus.BackupDir)
	_, err = f.KubeCli.CoreV1().Pods(e2eNamespace).Create(checkpod)
	Expect(err).NotTo(HaveOccurred())
	checkpods = append(checkpods, *checkpod)
	err = wait.Poll(5*time.Second, 30*time.Second, func() (bool, error) {
		checkpod, err = f.KubeCli.CoreV1().Pods(e2eNamespace).Get(checkpod.GetName(), metav1.GetOptions{})
		if err != nil {
			return false, nil
		}
		if checkpod.Status.Phase != corev1.PodSucceeded {
			return false, nil
		}
		return true, nil
	})
	Expect(err).NotTo(HaveOccurred())

	By(fmt.Sprintf("When update full backup's Suspend field to true "))
	fullbackup, err = f.Cli.PingcapV1().TidbClusterBackups(e2eNamespace).Get(fullBakcupName, metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())
	suspend := true
	fullbackup.Spec.FullBackup.Suspend = &suspend
	fullbackup, err = f.Cli.PingcapV1().TidbClusterBackups(e2eNamespace).Update(fullbackup)
	Expect(err).NotTo(HaveOccurred())

	By(fmt.Sprintf("Then full backup will be paused"))
	suspendPoint := time.Now()
	time.Sleep(3 * time.Minute)
	fullbackup, err = f.Cli.PingcapV1().TidbClusterBackups(e2eNamespace).Get(fullBakcupName, metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())
	for _, backupTaskStatus := range fullbackup.Status.FullBackupStatus.BackupTaskStatuses {
		Expect(backupTaskStatus.BeginTime.Time.Before(suspendPoint.Add(time.Minute))).To(Equal(true))
	}

	By(fmt.Sprintf("When we create a physical backup 's backup job "))
	physicalBakcupName := fmt.Sprintf("%s-physical-backup", clusterName)
	physicalbackup := &v1.TidbClusterBackup{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: e2eNamespace,
			Name:      physicalBakcupName,
		},
		Spec: v1.TidbClusterBackupSpec{
			Cluster:        clusterName,
			BackupImage:    backupImage,
			PhysicalBackup: &v1.PhysicalBackupSpec{},
		},
	}
	physicalbackup, err = f.Cli.PingcapV1().TidbClusterBackups(e2eNamespace).Create(physicalbackup)
	Expect(err).NotTo(HaveOccurred())

	By(fmt.Sprintf("Then physical backup 's backup job completed successfully "))
	err = wait.Poll(5*time.Second, 10*time.Minute, func() (bool, error) {
		physicalbackup, err = f.Cli.PingcapV1().TidbClusterBackups(e2eNamespace).Get(physicalBakcupName, metav1.GetOptions{})
		if err != nil {
			return false, nil
		}
		backupTaskStatuses := physicalbackup.Status.PhysicalBackupStatus.BackupTaskStatuses
		for _, backupTaskStatus := range backupTaskStatuses {
			if backupTaskStatus.State != v1.BackupSuccess {
				return false, nil
			}
		}
		return true, nil
	})
	Expect(err).NotTo(HaveOccurred())

	By(fmt.Sprintf("Then we will check the backup data "))
	physicalbackup, err = f.Cli.PingcapV1().TidbClusterBackups(e2eNamespace).Get(physicalBakcupName, metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())
	for storeID, backupTaskStatus := range physicalbackup.Status.PhysicalBackupStatus.BackupTaskStatuses {
		checkpod := getCheckPod(backupTaskStatus.BackupNode, backupTaskStatus.BackupDir)
		_, err = f.KubeCli.CoreV1().Pods(e2eNamespace).Create(checkpod)
		checkpods = append(checkpods, *checkpod)
		Expect(err).NotTo(HaveOccurred())
		err = wait.Poll(5*time.Second, 30*time.Second, func() (bool, error) {
			checkpod, err = f.KubeCli.CoreV1().Pods(e2eNamespace).Get(checkpod.GetName(), metav1.GetOptions{})
			if err != nil {
				return false, nil
			}
			if checkpod.Status.Phase == corev1.PodSucceeded {
				return true, nil
			}
			return true, fmt.Errorf("the storeID: [%s] on node: [%s]'s backup data does not exist", storeID, backupTaskStatus.BackupNode)
		})
		Expect(err).NotTo(HaveOccurred())
	}

	deleteCluster(f, e2eNamespace, clusterName, done, pdCount+tikvCount)
}

func getCheckPod(node, path string) *corev1.Pod {
	mountPath := "/var/backup"
	checkDir := filepath.Join(mountPath, filepath.Base(path))
	image, found := os.LookupEnv("OPERATOR_IMAGE")
	if !found {
		image = "localhost:5000/pingcap/tidb-operator"
	}
	return &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: e2eNamespace,
			Name:      fmt.Sprintf("backup-checker-%s", time.Now().Format("20060102150405")),
		},
		Spec: corev1.PodSpec{
			NodeName:      node,
			RestartPolicy: corev1.RestartPolicyNever,
			Volumes: []corev1.Volume{
				{
					Name: "backupdir",
					VolumeSource: corev1.VolumeSource{
						HostPath: &corev1.HostPathVolumeSource{
							Path: filepath.Dir(path),
						},
					},
				},
			},
			Containers: []corev1.Container{
				{
					Name:    "checker",
					Image:   image,
					Command: []string{"sh"},
					Args:    []string{"-c", fmt.Sprintf("[ '$(ls -A %s)' ] && exit 0 || exit 1", checkDir)},
					VolumeMounts: []corev1.VolumeMount{
						{
							Name:      "backupdir",
							MountPath: mountPath,
						},
					},
				},
			},
		},
	}
}