package impl

import (
	"context"
	"strconv"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/elliotchance/pie/v2"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

type bucketGrouperFunc func(ctx context.Context, bucket *bucketRecycler) (*bucketRecycler, bizerr.BizError)

type bucketRecycler struct {
	usingBuckets   map[entity.ID]*entity.Bucket
	uselessBuckets map[entity.ID]*entity.Bucket
}

func (b *bucketRecycler) markUseless(bucket *entity.Bucket) {
	b.uselessBuckets[bucket.ID] = bucket
	delete(b.usingBuckets, bucket.ID)
}

func (b *bucketRecycler) markUsing(bucket *entity.Bucket) {
	b.usingBuckets[bucket.ID] = bucket
	delete(b.uselessBuckets, bucket.ID)
}

type baseBucketWorker struct {
	bucketGroupers []bucketGrouperFunc
	bucketRepo     repo.BucketRepo
	backupRepo     backup.BackupRepo
	logBackupRepo  backup.LogBackupRepo

	coreCli grpcli.CoreGrpcClient
}

func (w *baseBucketWorker) getUselessBuckets(
	ctx context.Context, cloudProvider string, scope entity.BucketScope,
) (*bucketRecycler, bizerr.BizError) {
	logger := log.GetLogger(ctx)

	buckets, err := w.bucketRepo.ListByFilter(ctx, repo.BucketFilter{
		Provider: pointer.ToString(cloudProvider),
		Scope:    &scope,
		Statuses: []entity.BucketStatus{entity.BucketStatusCreating, entity.BucketStatusCreated},
	})
	if err != nil {
		logger.Error("recycle bucket, list bucket failed", zap.Error(err))
		return nil, err
	}
	if len(buckets) == 0 {
		logger.Info("not found bucket to delete")
		return &bucketRecycler{}, nil
	}

	logger.Info("recycle bucket, list bucket success", zap.Int("bucket_count", len(buckets)))

	bucketInfo := &bucketRecycler{
		usingBuckets:   make(map[entity.ID]*entity.Bucket, len(buckets)),
		uselessBuckets: make(map[entity.ID]*entity.Bucket, len(buckets)),
	}
	pie.Of(buckets).Each(func(b *entity.Bucket) {
		bucketInfo.usingBuckets[b.ID] = b
	})
	for _, filter := range w.bucketGroupers {
		var err bizerr.BizError
		bucketInfo, err = filter(ctx, bucketInfo)
		if err != nil {
			logger.Error("recycle bucket, filter bucket failed", zap.Error(err))
			return nil, err
		}
	}
	return bucketInfo, nil
}

func (w *baseBucketWorker) uselessBucketGroupByCreatedAt(
	ctx context.Context, bucketInfo *bucketRecycler,
) (*bucketRecycler, bizerr.BizError) {
	logger := log.GetLogger(ctx).Named("uselessBucketGroupByCreatedAt")
	logger.Info("recycle bucketInfo, filter by created_at")
	if len(bucketInfo.usingBuckets) == 0 {
		logger.Info("recycle bucketInfo, no using bucketInfo")
		return bucketInfo, nil
	}

	values := pie.Values(bucketInfo.usingBuckets)
	pie.Of(values).Each(func(b *entity.Bucket) {
		if b.Status == entity.BucketStatusCreating && time.Since(b.CreatedAt) > MaxBucketCreatingTime {
			bucketInfo.markUseless(b)
		}
	})
	logger.Info("recycle bucketInfo, filter by created_at success",
		zap.Any("using buckets", pie.Keys(bucketInfo.usingBuckets)),
		zap.Any("useless buckets", pie.Keys(bucketInfo.uselessBuckets)))
	return bucketInfo, nil
}

func (w *baseBucketWorker) uselessBucketGroupByBackups(
	ctx context.Context, bucketInfo *bucketRecycler,
) (*bucketRecycler, bizerr.BizError) {
	logger := log.GetLogger(ctx).Named("uselessBucketGroupByBackups")
	logger.Info("recycle bucketInfo, filter by backups")
	if len(bucketInfo.usingBuckets) == 0 {
		logger.Info("recycle bucketInfo, no using bucketInfo")
		return bucketInfo, nil
	}

	values := pie.Values(bucketInfo.usingBuckets)
	pie.Of(values).Each(func(b *entity.Bucket) {
		isEmpty, err := w.checkBucketIsEmpty(ctx, uint64(b.ID))
		if err != nil {
			logger.Warn("recycle bucketInfo, check bucket is empty failed", zap.Error(err))
			return
		}
		if isEmpty {
			bucketInfo.markUseless(b)
			return
		}
	})
	logger.Info("recycle bucketInfo, filter by backups success",
		zap.Any("using buckets", pie.Keys(bucketInfo.usingBuckets)),
		zap.Any("useless buckets", pie.Keys(bucketInfo.uselessBuckets)))
	return bucketInfo, nil
}

func (w *baseBucketWorker) checkBucketIsEmpty(
	ctx context.Context, bucketID uint64,
) (bool, bizerr.BizError) {
	backupCount, err := w.backupRepo.CountByFilter(
		ctx, backup.BackupFilter{CleanedUp: pointer.ToBool(false), BucketID: &bucketID})
	if err != nil {
		return false, err
	}
	logBackupCount, err2 := w.logBackupRepo.CountByFilter(
		ctx, backup.LogBackupFilter{CleanedUp: pointer.ToBool(false), BucketID: &bucketID})
	if err2 != nil {
		return false, err2
	}
	return backupCount+logBackupCount == 0, nil
}

func (w *baseBucketWorker) usingBucketGroupByCluster(
	ctx context.Context, bucketInfo *bucketRecycler,
) (*bucketRecycler, bizerr.BizError) {
	logger := log.GetLogger(ctx).Named("usingBucketGroupByCluster")
	logger.Info("recycle bucketInfo, filter by cluster")
	if len(bucketInfo.uselessBuckets) == 0 {
		logger.Info("recycle bucketInfo, no useless bucketInfo")
		return bucketInfo, nil
	}

	values := pie.Values(bucketInfo.uselessBuckets)
	pie.Of(values).Each(func(b *entity.Bucket) {
		projectID, _ := strconv.ParseUint(b.ProjectID, 10, 64)
		tenantID, _ := strconv.ParseUint(b.TenantID, 10, 64)
		// check if there are any not cleared(dedicated) cluster in this project
		listReq := &core.ListClustersReq{
			Filter: &core.ClusterFilter{
				TenantIds: []uint64{tenantID}, ProjectIds: []uint64{projectID},
				ExcludeStatuses: []core.ClusterStatus{core.ClusterStatus_CLUSTER_STATUS_CLEARED},
				IsDevTier:       pointer.ToBool(false),
			},
		}
		resp, listClusterErr := w.coreCli.ListClusters(ctx, listReq)
		if listClusterErr != nil {
			logger.Warn("list cluster in project failed",
				zap.Uint64("project_id", projectID), zap.Error(listClusterErr))
			bucketInfo.markUsing(b)
			return
		}
		if len(resp.Clusters) == 0 {
			return
		}
		anyMatch := pie.Of(pie.Values(resp.Clusters)).Any(func(c *core.Cluster) bool {
			return c.RegionName == b.Regions[0]
		})
		if anyMatch {
			bucketInfo.markUsing(b)
			return
		}
	})
	logger.Info("recycle bucketInfo, filter by cluster success",
		zap.Any("using buckets", pie.Keys(bucketInfo.usingBuckets)),
		zap.Any("useless buckets", pie.Keys(bucketInfo.uselessBuckets)))
	return bucketInfo, nil
}