// 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,
// 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.

// Test backup with exceeding GC safe point.

package main

import (
	"context"
	"flag"
	_ "net/http/pprof" //nolint:gosec // internal service
	"os"
	"strings"
	"time"

	"github.com/pingcap/kvproto/pkg/keyspacepb"
	"github.com/pingcap/log"
	"github.com/tidbcloud/aws-shared-provider/pkg/datacleaner/handle"
	"go.uber.org/zap"
)

const (
	defaultPdTimeout             = 10
	defaulTombstoneDelayDuration = time.Hour * 24 * 7
	defualtPerm                  = os.FileMode(0666)
	defaultMetaServerAddr        = "http://meta-store.tidb-admin.svc:9088"
	opTypsList                   = "dump_archive_ks, " +
		" dump_tombstone_ks, " +
		" archive_ks, " +
		" dump_pd_rules, " +
		" archive_pd_rules, " +
		" dump_region_labels, " +
		" archive_region_labels, " +
		" archive_etcd_path, " +
		" keyspace2tombstone, " +
		" clean_all_data, " +
		" clean_region_data"

	dump_archive_ks   = "dump_archive_ks"
	dump_tombstone_ks = "dump_tombstone_ks"
	archive_ks        = "archive_ks"

	dump_pd_rules      = "dump_pd_rules"
	dump_region_labels = "dump_region_labels"

	archive_pd_rules      = "archive_pd_rules"
	archive_region_labels = "archive_region_labels"

	archive_etcd_path  = "archive_etcd_path"
	keyspace2tombstone = "keyspace2tombstone"
	clean_all_data     = "clean_all_data"
	clean_region_data  = "clean_region_data"
)

var (
	ca                 = flag.String("ca", "", "CA certificate path for TLS connection")
	cert               = flag.String("cert", "", "certificate path for TLS connection")
	key                = flag.String("key", "", "private key path for TLS connection")
	dumpFilePath       = flag.String("dumpfile-ks", "dumpfile_ks.txt", "file to store archive keyspace list")
	dumpFilePdRulePath = flag.String(
		"dumpfile-pd-rules",
		"dumpfile_pd_rules.txt",
		"file to store all placement rules")
	dumpRegionLabelFilePath = flag.String(
		"dumpfile-region-labels",
		"dumpfile_region_labels.txt",
		"file to store all region labels")
	pdAddr         = flag.String("pd", "127.0.0.1:2379", "pd addrs")
	metaServerAddr = flag.String(
		"meta-server-addr",
		defaultMetaServerAddr,
		"meta server address for listing all keyspaces")
	opType = flag.String(
		"op",
		"",
		opTypsList)
	isRun                  = flag.Bool("isrun", false, "it's can real run operate")
	isSkipConfirm          = flag.Bool("skip-confirm", false, "it's skip confirm")
	pdTimeout              = flag.Int("pdTimeoutSec", defaultPdTimeout, "pd timeout (sec)")
	tombstoneDelayDuration = flag.Duration("delay", defaulTombstoneDelayDuration, "tombstone delay duration")
	keyspaceLimit          = flag.Int("limit", 0, "max number of keyspaces to clean per run, 0 means no limit")
	keyspaceInterval       = flag.Duration("keyspace-interval", time.Second, "wait duration between keyspaces")
	regionId               = flag.Uint64("region-id",
		0,
		"clean all the data of the keyspace covered by the region, keyspace must be in archived or tombstone state")
)

func vaildParam() bool {
	if len(*pdAddr) == 0 {
		log.Fatal("pd address is empty")
		return false
	}

	if len(*opType) == 0 {
		flag.Usage()
		return false
	}
	if *keyspaceInterval < 0 {
		log.Error("keyspace interval must not be negative")
		return false
	}
	return true
}

func hasOp(opList, target string) bool {
	for _, op := range splitOps(opList) {
		if op == target {
			return true
		}
	}
	return false
}

func main() {
	flag.Parse()
	logStartupParameters()
	ok := vaildParam()
	if !ok {
		return
	}

	defer func() {
		if globalClients != nil {
			_ = globalClients.Close()
		}
	}()

	for _, value := range splitOps(*opType) {
		switchOp(value, context.Background(), *isRun, *isSkipConfirm, *keyspaceInterval)
	}
}

func logStartupParameters() {
	log.Info("datacleaner start",
		zap.String("ca", *ca),
		zap.String("cert", *cert),
		zap.String("key", *key),
		zap.String("dumpfile-ks", *dumpFilePath),
		zap.String("dumpfile-pd-rules", *dumpFilePdRulePath),
		zap.String("dumpfile-region-labels", *dumpRegionLabelFilePath),
		zap.String("pd", *pdAddr),
		zap.String("meta-server-addr", *metaServerAddr),
		zap.String("op", *opType),
		zap.Strings("ops", splitOps(*opType)),
		zap.Bool("isrun", *isRun),
		zap.Bool("skip-confirm", *isSkipConfirm),
		zap.Int("pdTimeoutSec", *pdTimeout),
		zap.Duration("delay", *tombstoneDelayDuration),
		zap.Int("limit", *keyspaceLimit),
		zap.Duration("keyspace-interval", *keyspaceInterval),
		zap.Uint64("region-id", *regionId),
	)
}

func splitOps(opList string) []string {
	ops := strings.Split(opList, ",")
	values := make([]string, 0, len(ops))
	for _, op := range ops {
		if value := strings.TrimSpace(op); value != "" {
			values = append(values, value)
		}
	}
	return values
}

func switchOp(op string, ctx context.Context, isCanRun bool, isSkipConfirm bool, keyspaceInterval time.Duration) {
	switch op {
	case dump_archive_ks, dump_tombstone_ks: // Get archive keyspace id list
		dumpKeyspace(ctx, op)
	case archive_ks: // Delete Range by keyspace
		archiveKeyspace(ctx, isCanRun, isSkipConfirm, keyspaceInterval)
	case dump_pd_rules: // Dump all placement rules list
		dumpPDRules(ctx)
	case archive_pd_rules: // Archive placement rules by keyspace id.
		archiveLabels(ctx, isCanRun, isSkipConfirm, op, keyspaceInterval)
	case dump_region_labels: // Dump all region labels.
		dumpRegionLabels(ctx)
	case archive_region_labels: // Archive region labels by keyspace id.
		archiveLabels(ctx, isCanRun, isSkipConfirm, op, keyspaceInterval)
	case archive_etcd_path: // Archive etch path by keyspace id.
		archiveEtcdPath(ctx, isCanRun, isSkipConfirm, keyspaceInterval)
	case keyspace2tombstone: // Change keyspace state to tombstone
		changeKeyspace2Tombstone(ctx, isCanRun, isSkipConfirm, keyspaceInterval)
	case clean_all_data: // Clean all archived keyspace data
		cleanAllData(ctx, isCanRun, isSkipConfirm, keyspaceInterval)
	case clean_region_data: //  Clean region keyspaces data by region id
		cleanRegionKeyspaceData(ctx, *regionId)
	default:
		flag.Usage()
	}
}

func dumpKeyspace(ctx context.Context, op string) {
	dumpFile, err := os.OpenFile(*dumpFilePath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, defualtPerm)
	if err != nil {
		log.Fatal("Failed to open dumpFile", zap.Error(err))
	}

	var expectedState keyspacepb.KeyspaceState
	switch op {
	case dump_archive_ks:
		expectedState = keyspacepb.KeyspaceState_ARCHIVED
	case dump_tombstone_ks:
		expectedState = keyspacepb.KeyspaceState_TOMBSTONE
	}

	err = handle.DumpKeyspaceList(
		ctx,
		*metaServerAddr,
		dumpFile,
		*tombstoneDelayDuration,
		expectedState,
		*keyspaceLimit,
	)
	if err != nil {
		log.Fatal(err.Error())
	}
	err = dumpFile.Close()
	if err != nil {
		log.Fatal("Failed to close dumpFile", zap.Error(err))
	}
}

func dumpPDRules(ctx context.Context) {
	dumpFile, err := os.OpenFile(*dumpFilePdRulePath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, defualtPerm)
	if err != nil {
		log.Fatal("Failed to open dumpFile", zap.Error(err))
	}
	err = handle.DumpAllPlacementRules(ctx, getGlobalClients().GetPdContext(), dumpFile)
	if err != nil {
		log.Fatal(err.Error())
	}
	err = dumpFile.Close()
	if err != nil {
		log.Fatal("Failed to close dumpFile", zap.Error(err))
	}
}

func dumpRegionLabels(ctx context.Context) {
	dumpFile, err := os.OpenFile(*dumpRegionLabelFilePath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, defualtPerm)
	if err != nil {
		log.Fatal("Failed to open dumpFile", zap.Error(err))
	}

	err = handle.DumpAllRegionLabelRules(ctx, getGlobalClients().GetPdContext(), dumpFile)
	if err != nil {
		log.Fatal(err.Error())
	}

	err = dumpFile.Close()
	if err != nil {
		log.Fatal("Failed to close dumpFile", zap.Error(err))
	}
}

func archiveKeyspace(ctx context.Context, isCanRun, isSkipConfirm bool, keyspaceInterval time.Duration) {
	dumpFile, err := os.OpenFile(*dumpFilePath, os.O_RDWR, defualtPerm)
	if err != nil {
		log.Fatal("Failed to open dumpFile", zap.Error(err))
	}

	c := getGlobalClients()
	err = handle.LoadKeyspaceAndDeleteRange(ctx, dumpFile, c.PdClient, c.TikvClient, isCanRun, isSkipConfirm, keyspaceInterval)
	if err != nil {
		log.Fatal(err.Error())
	}

	err = dumpFile.Close()
	if err != nil {
		log.Fatal("Failed to close dumpFile", zap.Error(err))
	}
}

func archiveLabels(ctx context.Context, isCanRun, isSkipConfirm bool, op string, keyspaceInterval time.Duration) {
	dumpFile, err := os.OpenFile(*dumpFilePath, os.O_RDWR, defualtPerm)
	if err != nil {
		log.Fatal("Failed to close dumpFile", zap.Error(err))
	}

	var labelFilePath string
	switch op {
	case archive_pd_rules: // Archive placement rules by keyspace id.
		labelFilePath = *dumpFilePdRulePath
	case archive_region_labels: // Archive region labels by keyspace id.
		labelFilePath = *dumpRegionLabelFilePath
	default:
		log.Fatal("error op Type")
	}

	dumpFileLabels, err := os.OpenFile(labelFilePath, os.O_RDONLY, defualtPerm)
	if err != nil {
		log.Fatal("Failed to open dumpFileLabels", zap.Error(err), zap.String("labelFilePath", labelFilePath))
	}

	pdctx := getGlobalClients().GetPdContext()
	switch op {
	case archive_pd_rules: // Archive placement rules by keyspace id.
		err = handle.LoadPlacementRulesAndGC(
			ctx,
			dumpFileLabels,
			dumpFile,
			pdctx,
			isCanRun,
			isSkipConfirm,
			keyspaceInterval,
		)
	case archive_region_labels: // Archive region labels by keyspace id.
		err = handle.LoadRegionLablesAndGC(
			ctx,
			dumpFileLabels,
			dumpFile,
			pdctx,
			isCanRun,
			isSkipConfirm,
			keyspaceInterval,
		)
	default:
		log.Fatal("error op Type")
	}
	if err != nil {
		log.Fatal(err.Error())
	}

	err = dumpFile.Close()
	if err != nil {
		log.Fatal("Failed to close dumpFile", zap.Error(err))
	}
	err = dumpFileLabels.Close()
	if err != nil {
		log.Fatal("Failed to close dumpFileLabels", zap.Error(err), zap.String("labelFilePath", labelFilePath))
	}
}

func archiveEtcdPath(ctx context.Context, isCanRun, isSkipConfirm bool, keyspaceInterval time.Duration) {
	// create etcd client
	dumpFile, err := os.OpenFile(*dumpFilePath, os.O_RDWR, defualtPerm)
	if err != nil {
		log.Fatal("Failed to open dumpFile", zap.Error(err))
	}

	err = handle.LoadKeyspaceAndDeleteEtcdPath(
		ctx,
		dumpFile,
		getGlobalClients().PDEtcdClient,
		isCanRun,
		isSkipConfirm,
		getGlobalClients().MetaServiceGroupEtcdClient,
		keyspaceInterval,
	)
	if err != nil {
		log.Fatal(err.Error())
	}

	err = dumpFile.Close()
	if err != nil {
		log.Fatal("Failed to close dumpFile", zap.Error(err))
	}
}

func changeKeyspace2Tombstone(ctx context.Context, isCanRun, isSkipConfirm bool, keyspaceInterval time.Duration) {
	dumpFile, err := os.OpenFile(*dumpFilePath, os.O_RDWR, defualtPerm)
	if err != nil {
		log.Fatal("Failed to open dumpFile", zap.Error(err))
	}

	err = handle.LoadKeyspaceAndChange2Tombstone(
		ctx,
		dumpFile,
		getGlobalClients().PdClient,
		isCanRun,
		isSkipConfirm,
		keyspaceInterval,
	)
	if err != nil {
		log.Fatal(err.Error())
	}

	err = dumpFile.Close()
	if err != nil {
		log.Fatal("Failed to close dumpFile", zap.Error(err))
	}
}

func cleanAllData(ctx context.Context, isCanRun, isSkipConfirm bool, keyspaceInterval time.Duration) {
	c := getGlobalClients()
	err := handle.CleanAllArchiveKeyspaceDataWithInterval(
		ctx,
		c.PdClient,
		c.PDEtcdClient,
		c.TikvClient,
		c.GetPdContext(),
		isCanRun,
		isSkipConfirm,
		*tombstoneDelayDuration,
		c.MetaServiceGroupEtcdClient,
		*metaServerAddr,
		*keyspaceLimit,
		keyspaceInterval,
	)
	if err != nil {
		log.Fatal(err.Error())
	}
}

func cleanRegionKeyspaceData(ctx context.Context, regionId uint64) {
	c := getGlobalClients()
	err := handle.CleanTxnRegionKeyspaceData(ctx, c.PdClient, c.HttpPdClient, c.TikvClient, regionId, *isRun)
	if err != nil {
		log.Fatal(err.Error())
	}
}