package controller

import (
	"fmt"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/go-openapi/strfmt"
	"github.com/pingcap/log"
	"github.com/pingcap/tep/pkg/apis/apiserver"
	"github.com/pingcap/tep/pkg/operator"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/pingcap/tidb-operator/pkg/label"
	"go.uber.org/zap"
	batchv1 "k8s.io/api/batch/v1"
	batchv1beta1 "k8s.io/api/batch/v1beta1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
)

type BackupState string

const (
	BackupFailedState    BackupState = "failed"
	BackupRunningState   BackupState = "running"
	BackupSucceededState BackupState = "succeeded"
)

func splitDSN(dsn string) (string, string, string) {
	var service, password, user string
	user = strings.Split(dsn, ":")[0]
	password = strings.Split(strings.Split(dsn, "@")[0], ":")[1]
	service = strings.Split(strings.Split(dsn, "(")[1], ":")[0]
	return user, password, service
}

func createEnv(backup *apiserver.MetadataBackupTemplate) []corev1.EnvVar {
	var endpoint string
	if strings.HasPrefix(backup.S3.Endpoint, "http://") {
		endpoint = backup.S3.Endpoint
	} else {
		endpoint = fmt.Sprintf("http://%s", backup.S3.Endpoint)
	}
	return []corev1.EnvVar{
		{
			Name:  "AWS_ACCESS_KEY_ID",
			Value: backup.S3.AccessKey,
		},
		{
			Name:  "S3_ENDPOINT",
			Value: endpoint,
		},
		{
			Name:  "AWS_SECRET_ACCESS_KEY",
			Value: backup.S3.SecretKey,
		},
		{
			Name:  "S3_PROVIDER",
			Value: string(backup.S3.Provider),
		},
		{
			Name:  "AWS_REGION",
			Value: backup.S3.Region,
		},
		{
			Name:  "AWS_ACL",
			Value: backup.S3.ACL,
		},
		{
			Name:  "AWS_STORAGE_CLASS",
			Value: backup.S3.StorageClass,
		},
		{
			Name: "Name",
			ValueFrom: &corev1.EnvVarSource{
				FieldRef: &corev1.ObjectFieldSelector{
					FieldPath: "metadata.name",
				},
			},
		},
	}
}

func assembleS3StorageProvider(envList []corev1.EnvVar) *apiserver.S3StorageProvider {
	s3StorageProvider := &apiserver.S3StorageProvider{
		Bucket: operator.MetadataBucketName,
	}
	for _, env := range envList {
		if env.Name == "S3_ENDPOINT" {
			s3StorageProvider.Endpoint = env.Value
		}
		if env.Name == "AWS_REGION" {
			s3StorageProvider.Region = env.Value
		}
		if env.Name == "S3_PROVIDER" {
			s3StorageProvider.Provider = env.Value
		}
	}
	return s3StorageProvider
}

func getStatus(job *batchv1.Job) BackupState {
	// Check the status in conditions one by one,
	// if the status other than complete or failed is running
	status := BackupRunningState
	for _, condition := range job.Status.Conditions {
		if condition.Type == batchv1.JobComplete {
			status = BackupSucceededState
			break
		} else if condition.Type == batchv1.JobFailed {
			status = BackupFailedState
			break
		}
	}
	return status
}

func createCommand(namespace string, dsn string, backup *apiserver.MetadataBackupTemplate) []string {
	var metadataService, password, user, commandLine string
	user, password, metadataService = splitDSN(dsn)
	if password == "" {
		commandLine = fmt.Sprintf("/entrypoint.sh backup --namespace=%s --metadataservice=%s --storageType=%s --user=%s",
			namespace,
			metadataService,
			backup.StorageType,
			user)
	} else {
		commandLine = fmt.Sprintf("/entrypoint.sh backup --namespace=%s --metadataservice=%s --storageType=%s --password=%s --user=%s",
			namespace,
			metadataService,
			backup.StorageType,
			password,
			user)
	}

	return []string{
		"bin/sh",
		"-c",
		commandLine,
	}
}

func (c *Controller) CreateMetadataBackupJob(backupImage, namespace, dsn string, backup *apiserver.MetadataBackup) (*batchv1.Job, error) {
	var jobName string
	var job *batchv1.Job
	var err error

	jobName = fmt.Sprintf("%s-%s", operator.MetadataBackupJob, time.Now().UTC().Format(operator.BackupTimeFormat))

	labelMap := map[string]string{
		label.ComponentLabelKey: operator.MetadataBackupComponentLabel,
		"method":                operator.BackupOperationManual,
	}

	env := createEnv(&backup.MetadataBackupTemplate)

	command := createCommand(namespace, dsn, &backup.MetadataBackupTemplate)

	jobSpec := c.GetJobSpec(backupImage, labelMap, command, env, nil, nil, operator.TepServiceAccountName)

	if job, err = c.CreateJob(jobName, namespace, labelMap, *jobSpec); err != nil {
		msg := fmt.Errorf("failed to create job, namespace: %s, name: %s, err %v", namespace, jobName, err)
		return nil, msg
	}
	return job, nil
}

func (c *Controller) CreateMetadataRestoreJob(backupImage, namespace, dsn string, restore *apiserver.MetadataRestore) (*batchv1.Job, error) {
	var metadataService, password, user, commandLine, jobName, backupName, commitTs, backupPath string
	var err error
	var backupJob, job *batchv1.Job
	var ok bool

	backupName = restore.Backup

	if backupJob, err = c.GetJob(backupName, namespace); err != nil {
		msg := fmt.Errorf("failed to get job, namespace: %s, name: %s, err %v", namespace, backupName, err)
		return nil, msg
	}

	if commitTs, ok = backupJob.Labels[operator.TimeStartedLabel]; !ok {
		msg := fmt.Errorf("failed to get commitTs from job labels job: %s", backupName)
		return nil, msg
	}

	backupPath = fmt.Sprintf("%s://%s-%s/backup-%s%s",
		v1alpha1.BackupStorageTypeS3,
		namespace,
		operator.MetadataTCName,
		commitTs,
		operator.DefaultArchiveExtention)

	jobName = fmt.Sprintf("%s-%s", operator.MetadataRestoreJob, time.Now().UTC().Format(operator.BackupTimeFormat))

	user, password, metadataService = splitDSN(dsn)

	labelMap := map[string]string{
		label.ComponentLabelKey: "metadata-restore",
	}

	// construct job resource and create it
	env := backupJob.Spec.Template.Spec.Containers[0].Env

	if password == "" {
		commandLine = fmt.Sprintf("/entrypoint.sh restore --namespace=%s --metadataservice=%s --user=%s --backuppath=%s",
			namespace,
			metadataService,
			user,
			backupPath)
	} else {
		commandLine = fmt.Sprintf("/entrypoint.sh restore --namespace=%s --metadataservice=%s --password=%s --user=%s --backuppath=%s",
			namespace,
			metadataService,
			password,
			user,
			backupPath)
	}

	// construct jobspec command
	command := []string{
		"bin/sh",
		"-c",
		commandLine,
	}

	jobSpec := c.GetJobSpec(backupImage, labelMap, command, env, nil, nil, operator.TepServiceAccountName)

	if job, err = c.CreateJob(jobName, namespace, labelMap, *jobSpec); err != nil {
		msg := fmt.Errorf("failed to create job, namespace: %s, name: %s, err %v", namespace, jobName, err)
		return nil, msg
	}
	return job, nil
}

func (c *Controller) AssembleMetadataBackup(job *batchv1.Job) (*apiserver.MetadataBackup, error) {
	var val string
	var err error
	var ok bool
	var startTime, endTime time.Time

	backup := &apiserver.MetadataBackup{}
	backup.Name = job.Name
	backup.ObjectMeta.Name = job.ObjectMeta.Name
	backup.ObjectMeta.Namespace = job.ObjectMeta.Namespace
	if val, ok = job.Labels[operator.TimeStartedLabel]; ok {
		if startTime, err = time.Parse(operator.BackupTimeFormat, val); err != nil {
			return nil, err
		}
		backup.StartTime = strfmt.DateTime(startTime).String()
	}
	if val, ok = job.Labels[operator.TimeCompletedLabel]; ok {
		if endTime, err = time.Parse(operator.BackupTimeFormat, val); err != nil {
			return nil, err
		}
		backup.EndTime = strfmt.DateTime(endTime).String()
	}
	if val, ok = job.Labels[operator.BackupSizeLabel]; ok {
		if backup.Size, err = strconv.ParseInt(val, 10, 64); err != nil {
			return nil, err
		}
	}
	if val, ok = job.Labels["method"]; ok {
		backup.Operation = val
	}
	backup.Status = string(getStatus(job))
	// FIXME: hard code storageType
	backup.StorageType = string(v1alpha1.BackupStorageTypeS3)
	backup.S3 = assembleS3StorageProvider(job.Spec.Template.Spec.Containers[0].Env)
	return backup, nil
}

func (c *Controller) AssembleMetadataBackupCronJob(backupImage, namespace, dsn string, backup *apiserver.MetadataScheduledBackup) *batchv1beta1.CronJob {
	var jobName, schedule string
	var cronJob *batchv1beta1.CronJob

	schedule = backup.Schedule
	jobName = operator.MetadataBackupCronJob

	labelMap := map[string]string{
		label.ComponentLabelKey:       operator.MetadataBackupComponentLabel,
		"method":                      operator.BackupOperationAuto,
		operator.MaxReservedTimeLabel: backup.MaxReservedTime,
	}

	env := createEnv(&backup.MetadataBackupTemplate)

	command := createCommand(namespace, dsn, &backup.MetadataBackupTemplate)

	jobSpec := c.GetJobSpec(backupImage, labelMap, command, env, nil, nil, operator.TepServiceAccountName)

	failedJobsHistoryLimit := operator.MaxInt32
	successfulJobsHistoryLimit := operator.MaxInt32

	cronJob = &batchv1beta1.CronJob{
		ObjectMeta: metav1.ObjectMeta{
			Name:      jobName,
			Namespace: namespace,
			Labels:    labelMap,
		},
		Spec: batchv1beta1.CronJobSpec{
			Schedule:                   schedule,
			ConcurrencyPolicy:          "Replace",
			FailedJobsHistoryLimit:     &failedJobsHistoryLimit,
			SuccessfulJobsHistoryLimit: &successfulJobsHistoryLimit,
			JobTemplate: batchv1beta1.JobTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: labelMap,
				},
				Spec: *jobSpec,
			},
		},
	}

	return cronJob
}

func (c *Controller) AssembleMetadataScheduledBackup(job *batchv1beta1.CronJob) *apiserver.MetadataScheduledBackup {
	// FIXME: hard code the storage type
	backup := &apiserver.MetadataScheduledBackup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      job.ObjectMeta.Name,
			Namespace: job.ObjectMeta.Namespace,
			UID:       job.ObjectMeta.UID,
		},
		Schedule:        job.Spec.Schedule,
		MaxReservedTime: job.Labels[operator.MaxReservedTimeLabel],
		MetadataBackupTemplate: apiserver.MetadataBackupTemplate{
			StorageType: string(v1alpha1.BackupStorageTypeS3),
			StorageProvider: apiserver.StorageProvider{
				S3: assembleS3StorageProvider(job.Spec.JobTemplate.Spec.Template.Spec.Containers[0].Env),
			},
		},
	}
	return backup
}

// start a scheduled task, periodically delete expired backup task,
// as well as the remote backup data.
func (c *Controller) DeleteBackupViaReservedTime() {
	log.Info("clean up scheduled metadata backup begin", zap.String("start time", time.Now().Format(time.RFC3339)))
	defer log.Info("clean up scheduled metadata backup end", zap.String("end time", time.Now().Format(time.RFC3339)))

	var jobList *batchv1.JobList
	var job batchv1.Job
	var err error
	var reserved, namespace string
	var startTime time.Time
	var reservedTime time.Duration
	var ok bool
	namespace = os.Getenv("NAMESPACE")
	listOpts := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			map[string]string{
				label.ComponentLabelKey: operator.MetadataBackupComponentLabel,
			},
		).String(),
	}
	if jobList, err = c.ListJobs(namespace, listOpts); err != nil {
		log.Error("failed to get job list", zap.Error(err))
		return
	}
	for _, job = range jobList.Items {
		startTime = job.CreationTimestamp.Time
		if getStatus(&job) == BackupRunningState {
			// only tasks with results will be cleaned up
			continue
		}
		if reserved, ok = job.Labels[operator.MaxReservedTimeLabel]; ok {
			if reservedTime, err = time.ParseDuration(reserved); err != nil {
				log.Error("failed to parse duration ", zap.String("duration", reserved), zap.Error(err))
				continue
			}
			if time.Now().After(startTime.Add(reservedTime)) {
				// delete task as well as remote backup file if task is expired
				if err = c.DeleteBackupJob(job.Name, job.Namespace); err != nil {
					log.Error("failed to delete backup job ", zap.Error(err))
					continue
				}
			}
		}
	}

	listOpts = metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			map[string]string{
				label.ComponentLabelKey: operator.MetadataCleanComponentLabel,
			},
		).String(),
	}
	if jobList, err = c.ListJobs(namespace, listOpts); err != nil {
		log.Error("failed to get job list", zap.Error(err))
		return
	}

	// delete completed clean job
	for _, job = range jobList.Items {
		if getStatus(&job) != BackupRunningState {
			// only tasks with results will be cleaned up
			if err = c.DeleteJob(job.Name, job.Namespace); err != nil {
				continue
			}
		}
	}

	return
}

func (c *Controller) CreateMetadataCleanJob(backupName, namespace string) (*batchv1.Job, error) {
	var backupImage, commandLine, jobName, commitTs, backupPath string
	var err error
	var backupJob, job *batchv1.Job
	var ok bool

	if backupJob, err = c.GetJob(backupName, namespace); err != nil {
		msg := fmt.Errorf("failed to get job, namespace: %s, name: %s, err %v", namespace, backupName, err)
		return nil, msg
	}

	// If the backup is not successful, you do not need to delete the remote file.
	if getStatus(backupJob) != BackupSucceededState {
		return nil, nil
	}

	if commitTs, ok = backupJob.Labels[operator.TimeStartedLabel]; !ok {
		msg := fmt.Errorf("failed to get commitTs from job labels job: %s", backupName)
		return nil, msg
	}

	backupPath = fmt.Sprintf("%s://%s-%s/backup-%s%s",
		v1alpha1.BackupStorageTypeS3,
		namespace,
		operator.MetadataTCName,
		commitTs,
		operator.DefaultArchiveExtention)

	jobName = fmt.Sprintf("%s-%s", operator.MetadataCleanJob, backupName)

	labelMap := map[string]string{
		label.ComponentLabelKey: operator.MetadataCleanComponentLabel,
	}

	// construct job resource and create it
	env := backupJob.Spec.Template.Spec.Containers[0].Env
	backupImage = backupJob.Spec.Template.Spec.Containers[0].Image

	commandLine = fmt.Sprintf("/entrypoint.sh clean --namespace=%s --backupPath=%s",
		namespace,
		backupPath)

	// construct jobspec command
	command := []string{
		"bin/sh",
		"-c",
		commandLine,
	}

	jobSpec := c.GetJobSpec(backupImage, labelMap, command, env, nil, nil, operator.TepServiceAccountName)

	if job, err = c.CreateJob(jobName, namespace, labelMap, *jobSpec); err != nil {
		msg := fmt.Errorf("failed to create job, namespace: %s, name: %s, err %v", namespace, jobName, err)
		return nil, msg
	}
	return job, nil
}

// delelte backup job should delete remote file as well as job resource
func (c *Controller) DeleteBackupJob(name, namespace string) error {
	var err error
	if _, err = c.CreateMetadataCleanJob(name, namespace); err != nil {
		return err
	}
	return c.DeleteJob(name, namespace)
}

func (c *Controller) GetSecretKeyViaEnv(envList []corev1.EnvVar) (string, string) {
	var accessKey, secretKey string
	for _, env := range envList {
		if env.Name == "AWS_ACCESS_KEY_ID" {
			accessKey = env.Value
		}
		if env.Name == "AWS_SECRET_ACCESS_KEY" {
			secretKey = env.Value
		}
	}
	return accessKey, secretKey
}