// Copyright 2019 PingCAP, Inc.
//
// 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,
// See the License for the specific language governing permissions and
// limitations under the License.

package backup

import (
	"database/sql"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/pingcap/tep/cmd/tep-metadata-backup-manager/app/constants"
	"github.com/pingcap/tep/cmd/tep-metadata-backup-manager/app/util"
	"github.com/pingcap/tep/pkg/operator"
	batchv1 "k8s.io/api/batch/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/util/retry"
	glog "k8s.io/klog"
)

// BackupManager mainly used to manage backup related work
type BackupManager struct {
	kubernetes.Interface
	BackupOpts
}

// NewBackupManager return a BackupManager
func NewBackupManager(kubeCli kubernetes.Interface, backupOpts BackupOpts) *BackupManager {
	return &BackupManager{
		kubeCli,
		backupOpts,
	}
}

// ProcessBackup used to process the backup logic
func (bm *BackupManager) ProcessBackup() error {
	var db *sql.DB
	var err error
	err = wait.PollImmediate(constants.PollInterval, constants.CheckTimeout, func() (done bool, err error) {
		db, err = util.OpenDB(bm.getDSN(constants.TidbMetaDB))
		if err != nil {
			glog.Warningf("can't open connection to tidb cluster %s, err: %v", bm, err)
			return false, nil
		}

		if err := db.Ping(); err != nil {
			glog.Warningf("can't connect to tidb cluster %s, err: %s", bm, err)
			return false, nil
		}
		return true, nil
	})

	if err != nil {
		glog.Errorf("cluster %s connect failed, err: %s", bm, err)
		return err
	}

	defer db.Close()
	return bm.performBackup(db)
}

func (bm *BackupManager) performBackup(db *sql.DB) error {
	started := time.Now()
	commitTs := started.UTC().Format(operator.BackupTimeFormat)

	backupFullPath, err := bm.dumpTidbClusterData(commitTs)
	if err != nil {
		glog.Errorf("dump cluster %s data failed, err: %s", bm, err)
		return err
	}
	glog.Infof("dump cluster %s data to %s success", bm, backupFullPath)

	// TODO: Concurrent get file size and upload backup data to speed up processing time
	archiveBackupPath := backupFullPath + operator.DefaultArchiveExtention
	err = archiveBackupData(backupFullPath, archiveBackupPath)
	if err != nil {
		glog.Errorf("archive cluster %s backup data %s failed, err: %s", bm, archiveBackupPath, err)
		return err
	}
	glog.Infof("archive cluster %s backup data %s success", bm, archiveBackupPath)

	size, err := getBackupSize(archiveBackupPath)
	if err != nil {
		glog.Errorf("get cluster %s archived backup file %s size %d failed, err: %s", bm, archiveBackupPath, size, err)
		return err
	}
	glog.Infof("get cluster %s archived backup file %s size %d success", bm, archiveBackupPath, size)

	remotePath := strings.TrimPrefix(archiveBackupPath, operator.BackupRootPath+"/")
	bucketURI := bm.getDestBucketURI(remotePath)
	err = bm.backupDataToRemote(archiveBackupPath, bucketURI)
	if err != nil {
		glog.Errorf("backup cluster %s data to %s failed, err: %s", bm, bm.StorageType, err)
		return err
	}

	finish := time.Now()

	glog.Infof("commitTs is %s backupPath is %s", commitTs, bucketURI)

	err = retry.RetryOnConflict(retry.DefaultBackoff, func() error {
		var retryErr error
		var job *batchv1.Job
		var jobName string
		var nameSlice []string
		nameSlice = strings.Split(os.Getenv("Name"), "-")
		jobName = strings.Join(nameSlice[:len(nameSlice)-1], "-")

		if job, retryErr = bm.BatchV1().Jobs(bm.Namespace).Get(jobName, metav1.GetOptions{}); retryErr != nil {
			return retryErr
		}
		job.Labels[operator.TimeStartedLabel] = commitTs
		job.Labels[operator.TimeCompletedLabel] = finish.UTC().Format(operator.BackupTimeFormat)
		job.Labels[operator.BackupSizeLabel] = strconv.FormatInt(size, 10)
		if job, retryErr = bm.BatchV1().Jobs(bm.Namespace).Update(job); retryErr != nil {
			return retryErr
		}
		return nil
	})

	if err != nil {
		glog.Errorf("failed to update job %s label, err: %s", bm, err)
		return err
	}

	glog.Infof("backup cluster %s data to %s success", bm, bm.StorageType)

	return nil
}

// ProcessCleanBackup used to clean the specific backup
func (bm *BackupManager) ProcessCleanBackup() error {
	err := bm.cleanRemoteBackupData(bm.BackupPath)
	if err != nil {
		glog.Errorf("clean cluster %s backup %s failed, err: %s", bm, bm.BackupPath, err)
		return err
	}

	glog.Infof("clean cluster %s backup %s success", bm, bm.BackupPath)
	return nil
}