package controller

import (
	"fmt"
	"strings"
	"time"

	"github.com/go-openapi/strfmt"
	"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"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

var (
	backupJobLabels  = map[string]string{"app.kubernetes.io/component": "scheduled-backup"}
	metabackupLabels = map[string]string{"app.kubernetes.io/component": "metadata-backup"}
)

func (c *Controller) CreateBackup(backup *v1alpha1.Backup, ns string) (*v1alpha1.Backup, error) {
	return c.clusterCli.PingcapV1alpha1().Backups(ns).Create(backup)
}

func (c *Controller) CreateRestore(restore *v1alpha1.Restore, ns string) (*v1alpha1.Restore, error) {
	return c.clusterCli.PingcapV1alpha1().Restores(ns).Create(restore)
}

func (c *Controller) CreateScheduledBackup(backupSchedule *v1alpha1.BackupSchedule, ns string) (*v1alpha1.BackupSchedule, error) {
	return c.clusterCli.PingcapV1alpha1().BackupSchedules(ns).Create(backupSchedule)
}

func (c *Controller) ListBackups(namespace string, opts metav1.ListOptions) (*v1alpha1.BackupList, error) {
	return c.clusterCli.PingcapV1alpha1().Backups(namespace).List(opts)
}

func (c *Controller) ListRestores(namespace string, opts metav1.ListOptions) (*v1alpha1.RestoreList, error) {
	return c.clusterCli.PingcapV1alpha1().Restores(namespace).List(opts)
}

func (c *Controller) GetRestore(name, namespace string) (*v1alpha1.Restore, error) {
	return c.clusterCli.PingcapV1alpha1().Restores(namespace).Get(name, metav1.GetOptions{})
}

func (c *Controller) GetScheduledBackup(name, namespace string) (*v1alpha1.BackupSchedule, error) {
	return c.clusterCli.PingcapV1alpha1().BackupSchedules(namespace).Get(name, metav1.GetOptions{})
}

func (c *Controller) GetRestoreStatus(restore *v1alpha1.Restore) RecoveryState {
	// Check the status in conditions one by one,
	// if the status other than complete or failed is running
	// See: https://github.com/pingcap/tidb-operator/blob/master/pkg/apis/pingcap.com/v1alpha1/types.go#L468
	status := RecoveryUnknownState
	for _, condition := range restore.Status.Conditions {
		if condition.Type == v1alpha1.RestoreComplete {
			status = RecoverySucceededState
			break
		} else if condition.Type == v1alpha1.RestoreFailed {
			status = RecoveryFailedState
			break
		} else if condition.Type == v1alpha1.RestoreRunning || condition.Type == v1alpha1.RestoreScheduled {
			status = RecoveryRunningState
		}
	}
	return status
}

func (c *Controller) GetBackup(name, namespace string) (*v1alpha1.Backup, error) {
	return c.clusterCli.PingcapV1alpha1().Backups(namespace).Get(name, metav1.GetOptions{})
}

func (c *Controller) UpdateScheduledBackup(namespace string, backupSchedule *v1alpha1.BackupSchedule) (*v1alpha1.BackupSchedule, error) {
	return c.clusterCli.PingcapV1alpha1().BackupSchedules(namespace).Update(backupSchedule)
}

func (c *Controller) DeleteBackup(name, namespace string) error {
	var err error
	if _, err = c.GetBackup(name, namespace); err != nil {
		return err
	}
	return c.clusterCli.PingcapV1alpha1().Backups(namespace).Delete(name, &metav1.DeleteOptions{})
}

func (c *Controller) DeleteAllBackups(namespace string) error {
	return c.clusterCli.PingcapV1alpha1().Backups(namespace).DeleteCollection(&metav1.DeleteOptions{}, metav1.ListOptions{})
}

func (c *Controller) DeleteScheduledBackup(name, namespace string) error {
	var err error
	if _, err = c.GetScheduledBackup(name, namespace); err != nil {
		return err
	}
	return c.clusterCli.PingcapV1alpha1().BackupSchedules(namespace).Delete(name, &metav1.DeleteOptions{})
}

func (c *Controller) DeleteAllRestores(namespace string) error {
	return c.clusterCli.PingcapV1alpha1().Restores(namespace).DeleteCollection(&metav1.DeleteOptions{}, metav1.ListOptions{})
}

func getTidbClusterEndpoint(instance, namespace string) string {
	return fmt.Sprintf("%s-tidb.%s", instance, namespace)
}

func getClusterFromEndpoint(endpoint string) string {
	return strings.Split(endpoint, "-")[0]
}

// FIXME: if the backupschedule is to support multiple backupspecs later,
// this logic should be modified.
func renderBackupTemplate(backup *v1alpha1.BackupSpec) *apiserver.TiDBClusterBackupTemplate {
	return &apiserver.TiDBClusterBackupTemplate{
		Cluster:          getClusterFromEndpoint(backup.From.Host),
		Type:             string(backup.Type),
		Method:           operator.BackupMethodLogic,
		StorageClassName: backup.StorageClassName,
		StorageSize:      backup.StorageSize,
		StorageProvider: apiserver.StorageProvider{
			S3: &apiserver.S3StorageProvider{
				Provider:     string(backup.S3.Provider),
				Region:       backup.S3.Region,
				Bucket:       backup.S3.Bucket,
				Endpoint:     backup.S3.Endpoint,
				StorageClass: backup.S3.StorageClass,
			},
		},
	}
}

func renderBackupSpec(secretName, namespace string, backup *apiserver.TiDBClusterBackupTemplate) *v1alpha1.BackupSpec {
	return &v1alpha1.BackupSpec{
		Type: v1alpha1.BackupType(backup.Type),
		From: v1alpha1.TiDBAccessConfig{
			Host:       getTidbClusterEndpoint(backup.Cluster, namespace),
			Port:       int32(operator.TiDBPort),
			User:       operator.TiDBSuperUser,
			SecretName: backup.TiDBSecretName,
		},
		StorageClassName: backup.StorageClassName,
		StorageSize:      backup.StorageSize,
		StorageProvider: v1alpha1.StorageProvider{
			S3: &v1alpha1.S3StorageProvider{
				Provider:     v1alpha1.S3StorageProviderTypeCeph,
				Region:       backup.S3.Region,
				Bucket:       backup.S3.Bucket,
				Endpoint:     backup.S3.Endpoint,
				SecretName:   secretName,
				StorageClass: backup.S3.StorageClass,
			},
		},
	}
}

func (c *Controller) AssembleBackup(secretName string, namespace string, backup *apiserver.TiDBClusterBackup) *v1alpha1.Backup {
	// backup CR name is required
	name := fmt.Sprintf("%s-%s-%s", operator.BackupNamePrefix, backup.Cluster, time.Now().Format(operator.BackupTimeFormat))
	return &v1alpha1.Backup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: namespace,
		},
		Spec: *renderBackupSpec(secretName, namespace, &backup.TiDBClusterBackupTemplate),
	}
}

func (c *Controller) GetBackupStatus(backup *v1alpha1.Backup) v1alpha1.BackupConditionType {
	// Check the status in conditions one by one,
	// if the status other than complete or failed is running
	// See: https://github.com/pingcap/tidb-operator/blob/master/pkg/apis/pingcap.com/v1alpha1/types.go#L346
	var status v1alpha1.BackupConditionType
	status = v1alpha1.BackupRunning
	for _, condition := range backup.Status.Conditions {
		if condition.Type == v1alpha1.BackupComplete {
			status = v1alpha1.BackupComplete
		} else if condition.Type == v1alpha1.BackupFailed {
			status = v1alpha1.BackupFailed
		}
	}
	return status
}

func (c *Controller) AssembleTiDBClusterBackup(backup *v1alpha1.Backup) *apiserver.TiDBClusterBackup {
	var status, operation string
	status = string(c.GetBackupStatus(backup))

	//The TiDB-Operator will give each backupCR a "tidb.pingcap.com/backup-schedule" label,
	// which can be used to determine whether it is an automatic backup or a manual backup.
	if _, ok := backup.GetLabels()[label.BackupScheduleLabelKey]; ok {
		operation = operator.BackupOperationAuto
	} else {
		operation = operator.BackupOperationManual
	}

	return &apiserver.TiDBClusterBackup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      backup.Name,
			Namespace: backup.Namespace,
			UID:       backup.UID,
		},
		StartTime: strfmt.DateTime(backup.Status.TimeStarted.Time).String(),
		EndTime:   strfmt.DateTime(backup.Status.TimeCompleted.Time).String(),
		Size:      backup.Status.BackupSize,
		Status:    status,
		TiDBClusterBackupTemplate: apiserver.TiDBClusterBackupTemplate{
			StorageClassName: backup.Spec.StorageClassName,
			StorageSize:      backup.Spec.StorageSize,
			Cluster:          getClusterFromEndpoint(backup.Spec.From.Host),
			Type:             string(backup.Spec.Type),
			Method:           operator.BackupMethodLogic,
			Operation:        operation,
			TiDBSecretName:   backup.Spec.From.SecretName,
			StorageProvider: apiserver.StorageProvider{
				S3: &apiserver.S3StorageProvider{
					Provider:     string(backup.Spec.S3.Provider),
					Region:       backup.Spec.S3.Region,
					Bucket:       backup.Spec.S3.Bucket,
					Endpoint:     backup.Spec.S3.Endpoint,
					StorageClass: backup.Spec.S3.StorageClass,
				},
			},
		},
	}
}

func (c *Controller) AssembleScheduledBackup(secretName string, namespace string, backup *apiserver.TiDBClusterScheduledBackup) *v1alpha1.BackupSchedule {
	name := fmt.Sprintf("%s", operator.ScheduledBackupName)

	return &v1alpha1.BackupSchedule{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: namespace,
		},
		Spec: v1alpha1.BackupScheduleSpec{
			Schedule:         backup.Schedule,
			MaxReservedTime:  &backup.MaxReservedTime,
			Pause:            *backup.Pause,
			StorageClassName: backup.StorageClassName,
			StorageSize:      backup.StorageSize,
			BackupTemplate:   *renderBackupSpec(secretName, namespace, &backup.TiDBClusterBackupTemplate),
		},
	}
}

func (c *Controller) AssembleTiDBClusterScheduledBackup(backup *v1alpha1.BackupSchedule) *apiserver.TiDBClusterScheduledBackup {
	return &apiserver.TiDBClusterScheduledBackup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      backup.Name,
			Namespace: backup.Namespace,
			UID:       backup.UID,
		},
		MaxReservedTime:           *backup.Spec.MaxReservedTime,
		Schedule:                  backup.Spec.Schedule,
		Pause:                     &backup.Spec.Pause,
		TiDBClusterBackupTemplate: *renderBackupTemplate(&backup.Spec.BackupTemplate),
	}
}

func (c *Controller) AssembleRestore(namespace string, backup *v1alpha1.Backup, restore *apiserver.TiDBClusterRestore) *v1alpha1.Restore {
	// restore CR name is required
	name := fmt.Sprintf("%s-%s-%s", operator.RestoreNamePrefix, restore.Cluster, time.Now().Format(operator.BackupTimeFormat))
	storageProvider := backup.Spec.StorageProvider
	// FIXME: only support s3 storage provider now
	storageProvider.S3.Path = backup.Status.BackupPath
	return &v1alpha1.Restore{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: namespace,
		},
		Spec: v1alpha1.RestoreSpec{
			To: v1alpha1.TiDBAccessConfig{
				Host:       getTidbClusterEndpoint(restore.Cluster, namespace),
				Port:       int32(operator.TiDBPort),
				User:       operator.TiDBSuperUser,
				SecretName: restore.SecretName,
			},
			Type:             backup.Spec.Type,
			StorageClassName: restore.StorageClassName,
			StorageSize:      restore.StorageSize,
			StorageProvider:  storageProvider,
		},
	}
}

func (c *Controller) AssembleTiDBClusterRestore(restore *v1alpha1.Restore) *apiserver.TiDBClusterRestore {
	return &apiserver.TiDBClusterRestore{
		ObjectMeta: metav1.ObjectMeta{
			Name:      restore.Name,
			Namespace: restore.Namespace,
			UID:       restore.UID,
		},
		StartTime: strfmt.DateTime(restore.Status.TimeStarted.Time).String(),
		EndTime:   strfmt.DateTime(restore.Status.TimeCompleted.Time).String(),
		Status:    string(c.GetRestoreStatus(restore)),
		TiDBClusterRestoreTemplate: apiserver.TiDBClusterRestoreTemplate{
			Name:             restore.Name,
			StorageSize:      restore.Spec.StorageSize,
			StorageClassName: restore.Spec.StorageClassName,
			Cluster:          getClusterFromEndpoint(restore.Spec.To.Host),
		},
	}
}

// ReplaceS3Secret ... replace s3 secret whose data comes from s3-object namespace
func (c *Controller) ReplaceS3Secret(namespace string, accessKey string, secretKey string) error {
	// replace s3 secret for each tidb cluster
	s3SecretData := make(map[string][]byte)
	s3SecretData["access_key"] = []byte(accessKey)
	s3SecretData["secret_key"] = []byte(secretKey)
	var updateErr error

	if err := c.CreateSecret(operator.S3SecretName, namespace, s3SecretData); err != nil {
		if errors.IsAlreadyExists(err) {
			if updateErr = c.UpdateSecret(operator.S3SecretName, namespace, s3SecretData); updateErr != nil {
				return fmt.Errorf("failed to update s3 object secret, namespace: %s, name: %s, err: %v", namespace, operator.S3SecretName, updateErr)
			}
		} else {
			return fmt.Errorf("failed to create s3 object secret, namespace: %s, name: %s, err: %v", namespace, operator.S3SecretName, err)
		}
	}

	return nil
}

// Get AccessKey and SecretKey from secret
func (c *Controller) GetCredentialFromSecret(namespace string, secretName string) (string, string, error) {
	var err error
	var secret *corev1.Secret

	if secret, err = c.GetSecret(secretName, namespace); err != nil {
		return "", "", err
	}
	return string(secret.Data["access_key"]), string(secret.Data["secret_key"]), nil
}

// String converts map[string]string to a ListOptions.LabelSelector
func convertLabelSelector(labels map[string]string) string {
	var arr []string

	for k, v := range labels {
		arr = append(arr, fmt.Sprintf("%s=%s", k, v))
	}

	return strings.Join(arr, ",")
}