package impl

import (
	"context"
	"strconv"

	"github.com/AlekSi/pointer"
	"github.com/elliotchance/pie/v2"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/models/bucket"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"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/internal/config"
	"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 GCPBucketWorker struct {
	baseWorker *baseBucketWorker

	bucketScope entity.BucketScope
	bucketModel bucket.BucketModelV2
}

func NewGCPProjectBucketWorker(
	bucketRepo repo.BucketRepo,
	backupRepo backup.BackupRepo,
	logBackupRepo backup.LogBackupRepo,
	gcpBucketModel bucket.BucketModelV2,
	coreCli grpcli.CoreGrpcClient,
) *GCPBucketWorker {
	baseWorker := &baseBucketWorker{
		bucketRepo:    bucketRepo,
		backupRepo:    backupRepo,
		logBackupRepo: logBackupRepo,
		coreCli:       coreCli,
	}
	worker := &GCPBucketWorker{
		baseWorker:  baseWorker,
		bucketScope: entity.BucketScopeProject,
		bucketModel: gcpBucketModel,
	}

	baseWorker.bucketGroupers = []bucketGrouperFunc{
		baseWorker.uselessBucketGroupByCreatedAt,
		baseWorker.uselessBucketGroupByBackups,
		baseWorker.usingBucketGroupByCluster,
	}
	return worker
}

func NewGCPClusterBucketWorker(
	bucketRepo repo.BucketRepo,
	backupRepo backup.BackupRepo,
	logBackupRepo backup.LogBackupRepo,
	gcpBucketModel bucket.BucketModelV2,
	coreCli grpcli.CoreGrpcClient,
) *GCPBucketWorker {
	baseWorker := &baseBucketWorker{
		bucketRepo:    bucketRepo,
		backupRepo:    backupRepo,
		logBackupRepo: logBackupRepo,
		coreCli:       coreCli,
	}
	worker := &GCPBucketWorker{
		baseWorker:  baseWorker,
		bucketScope: entity.BucketScopeCluster,
		bucketModel: gcpBucketModel,
	}

	baseWorker.bucketGroupers = []bucketGrouperFunc{
		baseWorker.uselessBucketGroupByCreatedAt,
		baseWorker.uselessBucketGroupByBackups,
		worker.usingClusterScopeBucket,
	}
	return worker
}

func (w *GCPBucketWorker) RecycleBucket(ctx context.Context) bizerr.BizError {
	ctx = log.WithLogID(ctx)
	logger := log.GetLogger(ctx)

	bucketInfo, err := w.baseWorker.getUselessBuckets(ctx, "gcp", w.bucketScope)
	if err != nil {
		return err
	}
	if len(bucketInfo.uselessBuckets) == 0 {
		return nil
	}

	logger.Info("recycle bucket, unused bucket", zap.Any("bucket", bucketInfo))
	count := 0
	for _, b := range bucketInfo.uselessBuckets {
		count++
		if count > config.Global().BackupConfig.MaxBucketDeleteBatchCount {
			logger.Warn("recycle bucket, reach max bucket delete batch count",
				zap.Int("limit", config.Global().BackupConfig.MaxBucketDeleteBatchCount))
			return nil
		}
		req := &domain.DeleteBucketReqV2{
			BucketID:           pointer.ToUint64(uint64(b.ID)),
			InfraCloudProvider: pointer.To(domain.InfraProviderTypeGKE),
		}
		resp, err := w.bucketModel.DeleteBucket(ctx, req)
		if err != nil {
			logger.Warn("recycle bucket, delete bucket failed", zap.Any("bucket_id", b.ID), zap.Error(err))
			continue
		}
		logger.Info("recycle bucket, delete bucket success",
			zap.Any("bucket_id", b.ID), zap.Any("resp", resp))
	}
	return nil
}

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

	for _, bucket := range bucketInfo.uselessBuckets {
		if bucket.Scope != entity.BucketScopeCluster {
			return nil, bizerr.New(
				bizerr.ErrCodeUnknown, "bucket %d is not cluster scope, scope: %s", bucket.ID, bucket.Scope)
		}
		clusterID, err := strconv.ParseUint(bucket.ClusterID, 10, 64)
		if err != nil {
			logger.Warn("parse cluster id of bucket to uint error",
				zap.Uint64("bucket_id", bucket.ID.Uint64()),
				zap.String("cluster_id", bucket.ClusterID),
				zap.Error(err))
			continue
		}

		cluster, err := w.baseWorker.coreCli.GetAnyClusterByID(ctx, clusterID)
		if err != nil {
			logger.Warn("get cluster error, mark bucket to using",
				zap.Uint64("cluster_id", clusterID), zap.Uint64("bucket_id", bucket.ID.Uint64()), zap.Error(err))
			bucketInfo.markUsing(bucket)
			continue
		}
		if cluster.Status != core.ClusterStatus_CLUSTER_STATUS_CLEARED {
			bucketInfo.markUsing(bucket)
		}
	}

	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
}