package handle

import (
	"context"
	"fmt"
	"os"
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/kvproto/pkg/keyspacepb"
	"github.com/pingcap/log"
	"github.com/tidbcloud/aws-shared-provider/pkg/datacleaner/common"
	"github.com/tikv/client-go/v2/txnkv"
	pd "github.com/tikv/pd/client"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.uber.org/zap"
)

func ChangeKeyspace2Tombstone(ctx context.Context, pdClient pd.Client, record *common.KeyspaceRecord,
	isRun, isSkipConfirm bool) (successed bool, err error) {
	var confirmMsg string
	if isSkipConfirm {
		confirmMsg = "yes"
	} else {
		fmt.Printf("please confirm is't needs to change keyspace:%d state to tombstone.(yes/no)\n", record.KeyspaceId)
		_, _ = fmt.Scanln(&confirmMsg)
	}

	if confirmMsg == "yes" {
		if isRun {
			if record.ArchivedEtcdPath && record.ArchivedKeyspace &&
				record.ArchivedPdRule && record.ArchivedRegionLabel {
				log.Info("[REAL RUN]", zap.Uint32("keyspaceId", record.KeyspaceId))
				_, err = pdClient.UpdateKeyspaceState(ctx, record.KeyspaceId, keyspacepb.KeyspaceState_TOMBSTONE)
				if err != nil {
					log.Error("Failed to change keyspace to tombstone", zap.Error(err), zap.Uint32("keyspaceId", record.KeyspaceId))
				} else {
					successed = true
					log.Info("Successfully change keyspace to tombestone", zap.Uint32("keyspaceId", record.KeyspaceId))
				}
			} else {
				log.Info("Failed to change keyspace to tombstone", zap.Uint32("keyspaceId", record.KeyspaceId),
					zap.Bool("ArchivedKeyspace", record.ArchivedKeyspace),
					zap.Bool("ArchivedEtcdPath", record.ArchivedEtcdPath),
					zap.Bool("ArchivedPdRule", record.ArchivedPdRule),
					zap.Bool("ArchivedRegionLabel", record.ArchivedRegionLabel),
				)
				err = errors.Errorf("Failed to change keyspace to tombstone")
			}
		} else {
			log.Info("confirm yes but '-isrun' is not set, skip the operator.")
		}
	} else {
		log.Info("confirm: NO.")
	}
	return
}

func LoadKeyspaceAndChange2Tombstone(ctx context.Context, file *os.File,
	pdClient pd.Client, isRun, isSkipConfirm bool, keyspaceInterval time.Duration) error {
	recordsMap, err := common.CacheArchiveKeyspaceRecords(file)
	if err != nil {
		log.Error("Failed to load keyspace records from file", zap.Error(err))
		return err
	}

	failedKsCt := 0
	successedKsCt := 0
	totalKsArchive := 0
	for _, record := range recordsMap {
		totalKsArchive++
		successed, err := ChangeKeyspace2Tombstone(ctx, pdClient, record, isRun, isSkipConfirm)
		if err != nil {
			failedKsCt++
		}
		if successed {
			successedKsCt++
		}
		if totalKsArchive < len(recordsMap) {
			waitBetweenKeyspaces(keyspaceInterval)
		}
	}

	log.Info("Change keyspace to tombestone successed",
		zap.Int("totalKsArchive", totalKsArchive),
		zap.Int("failedKsCt", failedKsCt),
		zap.Int("successedKsCt", successedKsCt))
	return nil
}

func CleanAllArchiveKeyspaceData(ctx context.Context, pdClient pd.Client, etcdClient *clientv3.Client, tikvClient *txnkv.Client, pdctx *common.PdContext, isRun, isSkipConfirm bool, delayDuration time.Duration, keyspaceEtcdClients map[string]*clientv3.Client, metaServerAddr string, limit int) error {
	return CleanAllArchiveKeyspaceDataWithInterval(ctx, pdClient, etcdClient, tikvClient, pdctx,
		isRun, isSkipConfirm, delayDuration, keyspaceEtcdClients, metaServerAddr, limit, 0)
}

func CleanAllArchiveKeyspaceDataWithInterval(ctx context.Context, pdClient pd.Client, etcdClient *clientv3.Client, tikvClient *txnkv.Client, pdctx *common.PdContext, isRun, isSkipConfirm bool, delayDuration time.Duration, keyspaceEtcdClients map[string]*clientv3.Client, metaServerAddr string, limit int, keyspaceInterval time.Duration) error {
	// Step 1: get archive keyspace & get placement rules & get region label rules
	keyspaceRecords, err := common.GetKeyspace(ctx, metaServerAddr, delayDuration, keyspacepb.KeyspaceState_ARCHIVED, limit)
	if err != nil {
		return err
	}
	if len(keyspaceRecords) == 0 {
		log.Info("no archived keyspaces eligible for cleanup, skipping bulk PD calls")
		return nil
	}
	placementRules, err := common.GetAllPlacementRules(ctx, pdctx)
	if err != nil {
		return err
	}
	regionLabelRules, err := common.GetAllRegionLabelRules(ctx, pdctx)
	if err != nil {
		return err
	}

	// Step 2: filter placementRules / regionLabelRules and convert to map
	placementRulesM, err := common.PlacementRules2Map(placementRules)
	if err != nil {
		return err
	}
	regionLabelRulesM, err := common.RegionLabelRules2Map(regionLabelRules)
	if err != nil {
		return err
	}

	failedKsCt := 0
	successedKsCt := 0
	totalKsArchive := 0
	// Step 3: clean data and update keyspace stats to tombstone
	for i, keyspaceRecord := range keyspaceRecords {
		totalKsArchive++
		// delete range
		_, err := DeleteRange(ctx, pdClient, tikvClient, keyspaceRecord, isRun, isSkipConfirm)
		if err != nil {
			log.Warn("Failed to delete range",
				zap.Uint32("keyspaceId", keyspaceRecord.KeyspaceId), zap.Error(err))
		}
		// delete keyspace placement rule
		_, err = DeleteKeyspacePlacementRule(ctx, placementRulesM, pdctx, keyspaceRecord, isRun, isSkipConfirm)
		if err != nil {
			log.Warn("Failed to delete keyspace placement rule",
				zap.Uint32("keyspaceId", keyspaceRecord.KeyspaceId), zap.Error(err))
		}
		// delete keyspace region label
		_, err = DeleteKeyspaceRegionLabel(ctx, regionLabelRulesM, pdctx, keyspaceRecord, isRun, isSkipConfirm)
		if err != nil {
			log.Warn("Failed to delete keyspace region label",
				zap.Uint32("keyspaceId", keyspaceRecord.KeyspaceId), zap.Error(err))
		}
		// delete etcd path
		_, err = DeleteEtcdPath(ctx, etcdClient, keyspaceEtcdClients, keyspaceRecord, isRun, isSkipConfirm)
		if err != nil {
			log.Warn("Failed to delete keyspacee etcd path",
				zap.Uint32("keyspaceId", keyspaceRecord.KeyspaceId), zap.Error(err))
		}
		// update keyspace stats to tombstone
		successed, err := ChangeKeyspace2Tombstone(ctx, pdClient, keyspaceRecord, isRun, isSkipConfirm)
		if err != nil {
			log.Warn("Failed to change keyspace to tombstone",
				zap.Uint32("keyspaceId", keyspaceRecord.KeyspaceId), zap.Error(err))
			failedKsCt++
		}
		if successed {
			successedKsCt++
		}
		if i+1 < len(keyspaceRecords) {
			waitBetweenKeyspaces(keyspaceInterval)
		}
	}
	log.Warn("GC keyspace data end.",
		zap.Int("totalKsArchive", totalKsArchive),
		zap.Int("failedKsCt", failedKsCt),
		zap.Int("successedKsCt", successedKsCt))
	return nil
}

func waitBetweenKeyspaces(interval time.Duration) {
	if interval <= 0 {
		return
	}
	log.Info("waiting before processing next keyspace", zap.Duration("interval", interval))
	time.Sleep(interval)
}