package impl

import (
	"context"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/elliotchance/pie/v2"
	"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"
)

const (
	MaxBucketCreatingTime            = 60 * time.Minute
	MaxBucketReplicationDeletingHour = 24 * 3
)

type bucketReplicationGrouperFunc func(
	ctx context.Context, bucketReplication []entity.AWSBucketReplication,
) ([]entity.AWSBucketReplication, bizerr.BizError)

type AWSBucketWorker struct {
	baseWorker *baseBucketWorker

	bucketReplicationGroupers []bucketReplicationGrouperFunc

	bucketRepo            repo.BucketRepo
	bucketReplicationRepo repo.AWSBucketReplicationRepo
	backupRepo            backup.BackupRepo
	logBackupRepo         backup.LogBackupRepo

	bucketModelProxy bucket.BucketModelV2
	awsBucketManager bucket.AWSBucketManagerModel
}

func NewAWSBucketWorker(
	bucketRepo repo.BucketRepo,
	bucketReplicationRepo repo.AWSBucketReplicationRepo,
	backupRepo backup.BackupRepo,
	logBackupRepo backup.LogBackupRepo,
	bucketModelProxy bucket.BucketModelV2,
	coreCli grpcli.CoreGrpcClient,
	awsBucketManager bucket.AWSBucketManagerModel,
) *AWSBucketWorker {
	baseWorker := &baseBucketWorker{
		bucketRepo:    bucketRepo,
		backupRepo:    backupRepo,
		logBackupRepo: logBackupRepo,
		coreCli:       coreCli,
	}
	worker := &AWSBucketWorker{
		baseWorker:            baseWorker,
		bucketRepo:            bucketRepo,
		bucketReplicationRepo: bucketReplicationRepo,
		backupRepo:            backupRepo,
		logBackupRepo:         logBackupRepo,
		bucketModelProxy:      bucketModelProxy,
		awsBucketManager:      awsBucketManager,
	}

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

	worker.bucketReplicationGroupers = []bucketReplicationGrouperFunc{
		worker.uselessBucketReplicationGroupByBackups,
		worker.uselessBackupReplicationGroupByLastCleanupBackupTime,
	}
	return worker
}

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

	bucketInfo, err := w.baseWorker.getUselessBuckets(ctx, "aws", entity.BucketScopeProject)
	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.To(uint64(b.ID)),
			InfraCloudProvider: pointer.To(domain.InfraProviderTypeEKS),
		}
		resp, err := w.bucketModelProxy.DeleteBucket(ctx, req)
		if err != nil {
			logger.Warn("recycle bucket, delete bucket failed", 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 *AWSBucketWorker) usingBucketGroupByBucketReplication(
	ctx context.Context, bucketInfo *bucketRecycler,
) (*bucketRecycler, bizerr.BizError) {
	logger := log.GetLogger(ctx).Named("usingBucketGroupByBucketReplication")
	logger.Info("recycle bucketInfo, filter by bucket replications")
	if len(bucketInfo.usingBuckets) == 0 || len(bucketInfo.uselessBuckets) == 0 {
		logger.Info("recycle bucketInfo, no using bucketInfo or useless bucketInfo")
		return bucketInfo, nil
	}
	replications, err := w.bucketReplicationRepo.List(ctx,
		func(qs entity.AWSBucketReplicationQuerySet) entity.AWSBucketReplicationQuerySet {
			return qs.StatusNe(entity.AWSBucketReplicationStatusDeleted)
		})
	if err != nil {
		logger.Error("recycle bucketInfo, list bucket replications failed", zap.Error(err))
		return nil, err
	}
	if len(replications) == 0 {
		logger.Info("recycle bucketInfo, no bucket replication")
		return bucketInfo, nil
	}

	bucketWithReplication := make(map[entity.ID]struct{})
	pie.Of(replications).Each(func(r entity.AWSBucketReplication) {
		bucketWithReplication[r.SourceBucketID] = struct{}{}
		bucketWithReplication[r.TargetBucketID] = struct{}{}
	})

	values := pie.Values(bucketInfo.uselessBuckets)
	pie.Of(values).Each(func(b *entity.Bucket) {
		if _, ok := bucketWithReplication[b.ID]; ok {
			bucketInfo.markUsing(b)
			return
		}
	})
	logger.Info("recycle bucketInfo, filter by bucket replications success",
		zap.Any("using buckets", pie.Keys(bucketInfo.usingBuckets)),
		zap.Any("useless buckets", pie.Keys(bucketInfo.uselessBuckets)))
	return bucketInfo, nil
}

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

	replications, err := w.bucketReplicationRepo.List(ctx,
		func(qs entity.AWSBucketReplicationQuerySet) entity.AWSBucketReplicationQuerySet {
			return qs.StatusEq(entity.AWSBucketReplicationStatusDeleting)
		})
	if err != nil {
		logger.Error("recycle bucket replication, list bucket replications failed", zap.Error(err))
		return err
	}
	if len(replications) == 0 {
		return nil
	}

	for _, grouper := range w.bucketReplicationGroupers {
		var err bizerr.BizError
		replications, err = grouper(ctx, replications)
		if err != nil {
			logger.Error("recycle bucket replication, filter bucket replications failed", zap.Error(err))
			return err
		}
	}

	if len(replications) == 0 {
		logger.Info("recycle bucket replication, no bucket replication need to delete")
		return nil
	}

	pie.Of(replications).Each(func(r entity.AWSBucketReplication) {
		w.deleteBucketReplication(ctx, &r)
	})
	return nil
}

func (w *AWSBucketWorker) deleteBucketReplication(ctx context.Context, r *entity.AWSBucketReplication) {
	logger := log.GetLogger(ctx)
	sourceBucket, err := w.bucketRepo.Load(ctx, uint64(r.SourceBucketID))
	if err != nil {
		logger.Error("recycle bucket replication, load source bucket failed", zap.Error(err))
		return
	}
	if sourceBucket.Status == entity.BucketStatusDeleting ||
		sourceBucket.Status == entity.BucketStatusDeleted {
		logger.Info("recycle bucket replication, source bucket is deleting or deleted")
		err = w.markBucketReplicationDeleted(ctx, r)
		if err != nil {
			logger.Warn("recycle bucket replication, update bucket replication status failed", zap.Error(err))
			return
		}
		return
	}
	resp, err := w.awsBucketManager.DeleteBucketReplication(ctx, &domain.DeleteAWSBucketReplicationReq{
		AWSBucketReq: domain.AWSBucketReq{
			Name:         sourceBucket.Name,
			AWSAccountID: *sourceBucket.CloudAccountID,
			Region:       sourceBucket.Regions[0],
		},
		RuleID: pointer.To(r.Prefix),
	})
	if err != nil {
		logger.Error("recycle bucket replication, delete bucket replication failed",
			zap.Error(err), zap.Any("req", r))
		return
	}
	err = w.markBucketReplicationDeleted(ctx, r)
	if err != nil {
		logger.Warn("recycle bucket replication, update bucket replication status failed")
		return
	}
	logger.Info("recycle bucket replication, delete bucket replication success", zap.Any("resp", resp))
}

func (w *AWSBucketWorker) markBucketReplicationDeleted(
	ctx context.Context, r *entity.AWSBucketReplication,
) bizerr.BizError {
	r.Status = entity.AWSBucketReplicationStatusDeleted
	err := w.bucketReplicationRepo.Save(ctx, r)
	if err != nil {
		return err
	}
	return nil
}

func (w *AWSBucketWorker) uselessBucketReplicationGroupByBackups(
	ctx context.Context, bucketReplication []entity.AWSBucketReplication,
) ([]entity.AWSBucketReplication, bizerr.BizError) {
	logger := log.GetLogger(ctx).Named("uselessBucketReplicationGroupByBackups")
	logger.Info("recycle bucket replications, filter by backups")
	if len(bucketReplication) == 0 {
		logger.Info("recycle bucket replications, no using bucket replications")
		return bucketReplication, nil
	}

	bucketReplication = pie.Of(bucketReplication).Filter(func(r entity.AWSBucketReplication) bool {
		isEmpty, err := w.checkClusterBackupIsCleanup(ctx,
			pointer.To(uint64(r.SourceBucketID)), pointer.To(uint64(r.ClusterID)))
		if err != nil {
			logger.Warn("recycle bucket replication, check bucket is empty failed", zap.Error(err))
			return false
		}
		return isEmpty
	}).Result

	logger.Info("recycle bucket replications, filter by backups success",
		zap.Any("useless bucket replications", bucketReplication))
	return bucketReplication, nil
}

func (w *AWSBucketWorker) uselessBackupReplicationGroupByLastCleanupBackupTime(
	ctx context.Context, bucketReplication []entity.AWSBucketReplication,
) ([]entity.AWSBucketReplication, bizerr.BizError) {
	logger := log.GetLogger(ctx).Named("uselessBackupReplicationGroupByLastCleanupBackupTime")
	logger.Info("recycle bucket replications, filter by last cleanup backup time")
	if len(bucketReplication) == 0 {
		logger.Info("recycle bucket replications, no using bucket replications")
		return bucketReplication, nil
	}

	bucketReplication = pie.Of(bucketReplication).Filter(func(r entity.AWSBucketReplication) bool {
		isEmpty, err := w.checkClusterLastCleanupBackupTimeGtOneDay(ctx,
			pointer.To(uint64(r.SourceBucketID)), pointer.To(uint64(r.ClusterID)))
		if err != nil {
			logger.Warn("recycle bucket replication, check bucket is empty failed", zap.Error(err))
			return false
		}
		return isEmpty
	}).Result

	logger.Info("recycle bucket replications, filter by last cleanup backup time success",
		zap.Any("useless bucket replications", bucketReplication))
	return bucketReplication, nil
}

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

func (w *AWSBucketWorker) checkClusterLastCleanupBackupTimeGtOneDay(
	ctx context.Context, bucketID *uint64, clusterID *uint64,
) (bool, bizerr.BizError) {
	backupCount, err := w.backupRepo.Count(ctx, func(qs entity.BackupQuerySet) entity.BackupQuerySet {
		return qs.CleanedUpEq(true).
			BucketIDEq(entity.ID(*bucketID)).
			ClusterIDEq(entity.ID(*clusterID)).
			UpdatedAtGt(time.Now().Add(-time.Hour * MaxBucketReplicationDeletingHour))
	})
	if err != nil {
		return false, err
	}
	if backupCount > 0 {
		return false, nil
	}

	logBackupCount, err2 := w.logBackupRepo.Count(ctx, func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet {
		return qs.CleanedUpEq(true).
			BucketIDEq(entity.ID(*bucketID)).
			ClusterIDEq(entity.ID(*clusterID)).
			UpdatedAtGt(time.Now().Add(-time.Hour * MaxBucketReplicationDeletingHour))
	})
	if err2 != nil {
		return false, err2
	}
	if logBackupCount > 0 {
		return false, nil
	}
	return true, nil
}