/*
Copyright 2018 Pressinfra SRL

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package syncer

import (
	"fmt"
	"strings"

	"github.com/presslabs/controller-util/syncer"
	batch "k8s.io/api/batch/v1"
	core "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	logf "sigs.k8s.io/controller-runtime/pkg/log"

	api "github.com/bitpoke/mysql-operator/pkg/apis/mysql/v1alpha1"
	"github.com/bitpoke/mysql-operator/pkg/internal/mysqlbackup"
	"github.com/bitpoke/mysql-operator/pkg/internal/mysqlcluster"
	"github.com/bitpoke/mysql-operator/pkg/options"
)

var log = logf.Log.WithName("mysqlbackup.syncer.job")

type jobSyncer struct {
	job     *batch.Job
	backup  *mysqlbackup.MysqlBackup
	cluster *mysqlcluster.MysqlCluster

	opt *options.Options
}

// NewJobSyncer returns a syncer for backup jobs
func NewJobSyncer(c client.Client, s *runtime.Scheme, backup *mysqlbackup.MysqlBackup, cluster *mysqlcluster.MysqlCluster, opt *options.Options) syncer.Interface {
	obj := &batch.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      backup.GetNameForJob(),
			Namespace: backup.Namespace,
		},
	}

	sync := &jobSyncer{
		job:     obj,
		backup:  backup,
		cluster: cluster,
		opt:     opt,
	}

	return syncer.NewObjectSyncer("Job", backup.Unwrap(), obj, c, sync.SyncFn)
}

func (s *jobSyncer) SyncFn() error {
	if s.backup.Status.Completed {
		log.V(1).Info("backup already completed", "backup", s.backup, "key", s.cluster)
		// skip doing anything
		return syncer.ErrIgnore
	}

	if len(s.backup.GetBackupURL(s.cluster)) == 0 {
		return fmt.Errorf("can't get backupURL")
	}

	// check if job is already created an just update the status
	if !s.job.ObjectMeta.CreationTimestamp.IsZero() {
		s.updateStatus(s.job)
		return nil
	}

	s.job.Labels = map[string]string{
		"cluster": s.backup.Spec.ClusterName,
	}

	s.job.Spec.Template.Spec = s.ensurePodSpec(s.job.Spec.Template.Spec)
	return nil
}

// getBackupCandidate returns the hostname of the first not-lagged and
// replicating slave node, else returns the master node.
func (s *jobSyncer) getBackupCandidate() string {
	for _, node := range s.cluster.Status.Nodes {
		master := s.cluster.GetNodeCondition(node.Name, api.NodeConditionMaster)
		replicating := s.cluster.GetNodeCondition(node.Name, api.NodeConditionReplicating)
		lagged := s.cluster.GetNodeCondition(node.Name, api.NodeConditionLagged)

		if master == nil || replicating == nil || lagged == nil {
			continue
		}

		isMaster := master.Status == core.ConditionTrue
		isReplicating := replicating.Status == core.ConditionTrue
		isLagged := lagged.Status == core.ConditionTrue

		// select a node that is not master is replicating and is not lagged
		if !isMaster && isReplicating && !isLagged {
			return node.Name
		}
	}

	log.Info("no healthy slave node found so returns the master node", "master_node", s.cluster.GetMasterHost(),
		"key", s.cluster, "backup", s.backup)
	return s.cluster.GetMasterHost()
}

// nolint: gocyclo
func (s *jobSyncer) ensurePodSpec(in core.PodSpec) core.PodSpec {
	if len(in.Containers) == 0 {
		in.Containers = make([]core.Container, 1)
	}

	in.RestartPolicy = core.RestartPolicyNever
	in.ImagePullSecrets = []core.LocalObjectReference{
		{Name: s.opt.ImagePullSecretName},
	}

	in.Containers[0].Name = "backup"
	in.Containers[0].Image = s.cluster.GetSidecarImage()
	in.Containers[0].ImagePullPolicy = s.opt.ImagePullPolicy
	in.Containers[0].Args = []string{
		"take-backup-to",
		s.getBackupCandidate(),
		s.backup.GetBackupURL(s.cluster),
	}

	in.ImagePullSecrets = s.cluster.Spec.PodSpec.ImagePullSecrets
	in.ServiceAccountName = s.cluster.Spec.PodSpec.ServiceAccountName

	in.Affinity = s.cluster.Spec.PodSpec.BackupAffinity
	if s.cluster.Spec.PodSpec.BackupAffinity == nil {
		in.Affinity = s.cluster.Spec.PodSpec.Affinity
	}

	in.NodeSelector = s.cluster.Spec.PodSpec.BackupNodeSelector
	if s.cluster.Spec.PodSpec.BackupNodeSelector == nil {
		in.NodeSelector = s.cluster.Spec.PodSpec.NodeSelector
	}

	in.PriorityClassName = s.cluster.Spec.PodSpec.BackupPriorityClassName
	if len(s.cluster.Spec.PodSpec.BackupPriorityClassName) == 0 {
		in.PriorityClassName = s.cluster.Spec.PodSpec.PriorityClassName
	}

	in.Tolerations = s.cluster.Spec.PodSpec.BackupTolerations
	if s.cluster.Spec.PodSpec.BackupTolerations == nil {
		in.Tolerations = s.cluster.Spec.PodSpec.Tolerations
	}

	boolTrue := true
	in.Containers[0].Env = []core.EnvVar{
		{
			Name: "BACKUP_USER",
			ValueFrom: &core.EnvVarSource{
				SecretKeyRef: &core.SecretKeySelector{
					LocalObjectReference: core.LocalObjectReference{
						Name: s.cluster.GetNameForResource(mysqlcluster.Secret),
					},
					Key:      "BACKUP_USER",
					Optional: &boolTrue,
				},
			},
		},
		{
			Name: "BACKUP_PASSWORD",
			ValueFrom: &core.EnvVarSource{
				SecretKeyRef: &core.SecretKeySelector{
					LocalObjectReference: core.LocalObjectReference{
						Name: s.cluster.GetNameForResource(mysqlcluster.Secret),
					},
					Key:      "BACKUP_PASSWORD",
					Optional: &boolTrue,
				},
			},
		},
	}

	hasBackupCompressCommand := len(s.cluster.Spec.BackupCompressCommand) > 0
	hasBackupDecompressCommand := len(s.cluster.Spec.BackupDecompressCommand) > 0
	if hasBackupCompressCommand && hasBackupDecompressCommand {
		in.Containers[0].Env = append(in.Containers[0].Env, core.EnvVar{
			Name:  "BACKUP_COMPRESS_COMMAND",
			Value: strings.Join(s.cluster.Spec.BackupCompressCommand, " "),
		})
	} else if hasBackupCompressCommand {
		log.Info("backupDecompressCommand is not defined, falling back to gzip")
	}

	if len(s.cluster.Spec.RcloneExtraArgs) > 0 {
		in.Containers[0].Env = append(in.Containers[0].Env, core.EnvVar{
			Name:  "RCLONE_EXTRA_ARGS",
			Value: strings.Join(s.cluster.Spec.RcloneExtraArgs, " "),
		})
	}

	if len(s.backup.Spec.BackupSecretName) != 0 {
		in.Containers[0].EnvFrom = []core.EnvFromSource{
			{
				SecretRef: &core.SecretEnvSource{
					LocalObjectReference: core.LocalObjectReference{
						Name: s.backup.Spec.BackupSecretName,
					},
				},
			},
		}
	}
	return in
}

func (s *jobSyncer) updateStatus(job *batch.Job) {
	// check for completion condition
	if cond := jobCondition(batch.JobComplete, job); cond != nil {
		s.backup.UpdateStatusCondition(api.BackupComplete, cond.Status, cond.Reason, cond.Message)

		if cond.Status == core.ConditionTrue {
			s.backup.Status.Completed = true
		}
	}

	// check for failed condition
	if cond := jobCondition(batch.JobFailed, job); cond != nil {
		s.backup.UpdateStatusCondition(api.BackupFailed, cond.Status, cond.Reason, cond.Message)

		if cond.Status == core.ConditionTrue {
			s.backup.Status.Completed = true
		}
	}
}

func jobCondition(condType batch.JobConditionType, job *batch.Job) *batch.JobCondition {
	for _, c := range job.Status.Conditions {
		if c.Type == condType {
			return &c
		}
	}

	return nil
}
