package handle

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

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

const KeyspaceConfigKeyMetaServiceGroupID = "meta_service_group_id"

func DeleteEtcdPath(ctx context.Context, pdEtcdClient *clientv3.Client, etcdGroupClients map[string]*clientv3.Client, keyspaceRecord *common.KeyspaceRecord, isRun, isSkipConfirm bool) (successed bool, err error) {
	// support meta service group.
	keyspaceEtcdCli, err := getEtcdClientByKeyspaceMeta(keyspaceRecord, pdEtcdClient, etcdGroupClients)
	if err != nil {
		log.Error("get etcd client by keyspace meta failed", zap.Error(err), zap.Any("keyspace-meta", keyspaceRecord.KeyspaceMeta))
		return false, err
	}
	keyspaceId := keyspaceRecord.KeyspaceId
	key := common.GetKeyspceEtcdPath(keyspaceId)
	var confirmMsg string
	if isSkipConfirm {
		confirmMsg = "yes"
	} else {
		fmt.Printf("please confirm is't that etcd path:%s needs to be gc.(yes/no)\n", key)
		_, _ = fmt.Scanln(&confirmMsg)
	}

	if confirmMsg == "yes" {
		if isRun {
			log.Info("[REAL RUN]", zap.Uint32("keyspaceId", keyspaceId), zap.Any("etcd-addr", keyspaceEtcdCli.Endpoints()))
			_, err = keyspaceEtcdCli.Delete(ctx, key, clientv3.WithPrefix())
			if err != nil {
				log.Error("DeleteEtcdPath failed to do delete", zap.Error(err))
				return successed, err
			}
			keyspaceRecord.ArchivedEtcdPath = true
			successed = true
			log.Info("DeleteEtcdPath success.", zap.Uint32("keyspaceId", keyspaceId), zap.String("etcdPath", key))
		} else {
			log.Info("confirm yes but '-isrun' is not set, skip the operator.")
		}
	} else {
		log.Info("confirm: NO.")
	}

	return
}

func LoadKeyspaceAndDeleteEtcdPath(ctx context.Context, file *os.File, pdEtcdClient *clientv3.Client, isRun, isSkipConfirm bool, etcdGroupClients map[string]*clientv3.Client, 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 _, keyspaceRecord := range recordsMap {
		totalKsArchive++
		successed, err := DeleteEtcdPath(ctx, pdEtcdClient, etcdGroupClients, keyspaceRecord, isRun, isSkipConfirm)
		if err != nil {
			failedKsCt++
		}
		if successed {
			successedKsCt++
		}
		if totalKsArchive < len(recordsMap) {
			waitBetweenKeyspaces(keyspaceInterval)
		}
	}

	err = common.UpdateArchiveKeyspaceRecords(recordsMap, file)
	if err != nil {
		log.Error("Failed to update KeyspaceRecordsFile", zap.Error(err))
		return err
	}
	log.Info("GC etcd path end.",
		zap.Int("totalKsArchive", totalKsArchive),
		zap.Int("failedKsCt", failedKsCt),
		zap.Int("successedKsCt", successedKsCt))

	return nil
}

func getEtcdClientByKeyspaceMeta(keyspaceRecord *common.KeyspaceRecord, pdEtcdClient *clientv3.Client, keyspaceEtcdClients map[string]*clientv3.Client) (*clientv3.Client, error) {
	var resEtcdClient *clientv3.Client
	if keyspaceMetaServiceGroupID, ok := keyspaceRecord.KeyspaceMeta.Config[KeyspaceConfigKeyMetaServiceGroupID]; ok {
		if etcdClient, ok := keyspaceEtcdClients[keyspaceMetaServiceGroupID]; ok {
			resEtcdClient = etcdClient
		} else {
			return nil, fmt.Errorf("etcd client for keyspace %s with group id %s not found", keyspaceRecord.KeyspaceMeta, keyspaceMetaServiceGroupID)
		}
	} else {
		resEtcdClient = pdEtcdClient
	}

	if resEtcdClient == nil {
		return nil, fmt.Errorf("etcd client for keyspace %s not found", keyspaceRecord.KeyspaceMeta)
	}

	log.Info("archive keyspace etcd key,etcd client", zap.Any("etcd-client-in-map", resEtcdClient.Endpoints()), zap.Any("keyspace-meta", keyspaceRecord.KeyspaceMeta))
	return resEtcdClient, nil
}