// 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 (
	"fmt"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"

	"github.com/mholt/archiver"
	"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"
	glog "k8s.io/klog"
)

// BackupOpts contains the input arguments to the backup command
type BackupOpts struct {
	Namespace       string
	MetadataService string
	Password        string
	User            string
	StorageType     string
	BackupPath      string
}

func (bo *BackupOpts) String() string {
	return fmt.Sprintf("%s/%s", bo.Namespace, operator.MetadataTCName)
}

func (bo *BackupOpts) getBackupFullPath(commitTs string) string {
	return filepath.Join(operator.BackupRootPath, bo.getBackupRelativePath(commitTs))
}

func (bo *BackupOpts) getBackupRelativePath(commitTs string) string {
	backupName := fmt.Sprintf("backup-%s", commitTs)
	return fmt.Sprintf("%s-%s/%s", bo.Namespace, operator.MetadataTCName, backupName)
}

func (bo *BackupOpts) getDestBucketURI(remotePath string) string {
	return fmt.Sprintf("%s://%s", bo.StorageType, remotePath)
}

func (bo *BackupOpts) dumpTidbClusterData(commitTs string) (string, error) {
	bfPath := bo.getBackupFullPath(commitTs)
	err := util.EnsureDirectoryExist(bfPath)
	if err != nil {
		return "", err
	}
	args := []string{
		fmt.Sprintf("--outputdir=%s", bfPath),
		fmt.Sprintf("--host=%s", bo.MetadataService),
		"--port=3306",
		fmt.Sprintf("--user=%s", bo.User),
		fmt.Sprintf("--password=%s", bo.Password),
		"--database=tep",
	}

	output, err := exec.Command("/mydumper", args...).CombinedOutput()
	if err != nil {
		return bfPath, fmt.Errorf("cluster %s, execute mydumper command %v failed, output: %s, err: %v", bo, args, string(output), err)
	}
	return bfPath, nil
}

func (bo *BackupOpts) backupDataToRemote(source, bucketURI string) error {
	destBucket := util.NormalizeBucketURI(bucketURI)
	tmpDestBucket := fmt.Sprintf("%s.tmp", destBucket)
	// TODO: We may need to use exec.CommandContext to control timeouts.
	output, err := exec.Command("rclone", constants.RcloneConfigArg, "copyto", source, tmpDestBucket).CombinedOutput()
	if err != nil {
		return fmt.Errorf("cluster %s, execute rclone copyto command for upload backup data %s failed, output: %s, err: %v", bo, bucketURI, string(output), err)
	}

	glog.Infof("upload cluster %s backup data to %s successfully, now move it to permanent URL %s", bo, tmpDestBucket, destBucket)

	// the backup was a success
	// remove .tmp extension
	output, err = exec.Command("rclone", constants.RcloneConfigArg, "moveto", tmpDestBucket, destBucket).CombinedOutput()
	if err != nil {
		return fmt.Errorf("cluster %s, execute rclone moveto command failed, output: %s, err: %v", bo, string(output), err)
	}
	return nil
}

func (bo *BackupOpts) cleanRemoteBackupData(bucket string) error {
	destBucket := util.NormalizeBucketURI(bucket)
	output, err := exec.Command("rclone", constants.RcloneConfigArg, "deletefile", destBucket).CombinedOutput()
	if err != nil {
		return fmt.Errorf("cluster %s, execute rclone deletefile command failed, output: %s, err: %v", bo, string(output), err)
	}

	glog.Infof("cluster %s backup %s was deleted successfully", bo, bucket)
	return nil
}

func (bo *BackupOpts) getDSN(db string) string {
	return fmt.Sprintf("%s:%s@(%s:3306)/%s?charset=utf8", bo.User, bo.Password, bo.MetadataService, db)
}

// getBackupSize get the backup data size
func getBackupSize(backupPath string) (int64, error) {
	var size int64
	if exist := util.IsFileExist(backupPath); !exist {
		return size, fmt.Errorf("file %s does not exist or is not regular file", backupPath)
	}
	out, err := exec.Command("rclone", constants.RcloneConfigArg, "ls", backupPath).CombinedOutput()
	if err != nil {
		return size, fmt.Errorf("failed to get backup %s size, err: %v", backupPath, err)
	}
	sizeStr := strings.Fields(string(out))[0]
	size, err = strconv.ParseInt(sizeStr, 10, 64)
	if err != nil {
		return size, fmt.Errorf("failed to parse size string %s, err: %v", sizeStr, err)
	}
	return size, nil
}

// archiveBackupData archive backup data by destFile's extension name
func archiveBackupData(backupDir, destFile string) error {
	if exist := util.IsDirExist(backupDir); !exist {
		return fmt.Errorf("dir %s does not exist or is not a dir", backupDir)
	}
	destDir := filepath.Dir(destFile)
	if err := util.EnsureDirectoryExist(destDir); err != nil {
		return err
	}
	err := archiver.Archive([]string{backupDir}, destFile)
	if err != nil {
		return fmt.Errorf("archive backup data %s to %s failed, err: %v", backupDir, destFile, err)
	}
	return nil
}