package handle

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

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

func DeleteKeyspacePlacementRule(ctx context.Context,
	placementRules map[string]common.PlacementRules, pdctx *common.PdContext,
	record *common.KeyspaceRecord, isRun, isSkipConfirm bool) (successed bool, err error) {
	keyspaceId := fmt.Sprintf("%d", record.KeyspaceId)

	if rules, ok := placementRules[keyspaceId]; ok {
		log.Info("[BEGIN] gc placement rules", zap.String("keyspaceId", keyspaceId))
		for _, rule := range rules {
			if err = common.DeletePlacementRule(ctx, pdctx, rule, isRun, isSkipConfirm); err != nil {
				log.Error("Delete placement rule failed", zap.String("keyspaceId", keyspaceId),
					zap.String("placementRule", rule.ID), zap.Error(err))
				// Once the deletion fails, we should stop the deletion and try again in the next round of cleanup.
				break
			} else if isRun {
				log.Info("Delete placement rule success.", zap.String("keyspaceID", keyspaceId),
					zap.String("placementRule", rule.ID))
			}
		}
	} else {
		log.Info("Can't find placement rule for keyspce", zap.String("keyspaceID", keyspaceId))
	}

	if err == nil && isRun {
		record.ArchivedPdRule = true
		successed = true
	}
	return
}

func LoadPlacementRulesAndGC(ctx context.Context, placementRulesFile,
	archiveKeyspaceFile *os.File, pdctx *common.PdContext, isRun, isSkipConfirm bool, keyspaceInterval time.Duration) error {
	keyspaceRecords, err := common.CacheArchiveKeyspaceRecords(archiveKeyspaceFile)
	if err != nil {
		log.Error("Failed to cache archived keyspace records", zap.Error(err))
		return err
	}
	placementRules, err := common.CachePlacementRules(placementRulesFile)
	if err != nil {
		log.Error("Failed to cache archived placement rules", zap.Error(err))
		return err
	}

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

	err = common.UpdateArchiveKeyspaceRecords(keyspaceRecords, archiveKeyspaceFile)
	if err != nil {
		log.Fatal("Failed to update archived keyspace records", zap.Error(err))
	}

	log.Info("GC placement rules end.",
		zap.Int("totalKsArchive", totalKsArchive),
		zap.Int("failedKsCt", failedKsCt),
		zap.Int("successedKsCt", successedKsCt))

	return nil
}

func DumpAllPlacementRules(ctx context.Context, pdctx *common.PdContext, f *os.File) error {
	rules, err := common.GetAllPlacementRules(ctx, pdctx)
	if err != nil {
		return err
	}
	return rules.WriteTo(f)
}