package jobs

import (
	"context"
	"fmt"

	"github.com/tidbcloud/dataflow-service/app/container"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/pkg/blob"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"go.uber.org/zap"
)

func CleanBucketData(ctx context.Context, bucketID uint64) error {
	logger := log.GetLogger(ctx).Named("CleanBucketData").With(zap.Uint64("bucket id", bucketID))
	app := container.Global()
	bucket, err := app.BucketRepo.Load(ctx, bucketID)
	if err != nil {
		logger.Error("load bucket error", zap.Error(err))
		return err
	}
	// check if we can clean the bucket data
	if bucket.Status != entity.BucketStatusDeleting {
		logger.Error("can't delete bucket without deleting status",
			zap.String("bucket status", string(bucket.Status)))
		return fmt.Errorf("can't delete bucket with status %s", bucket.Status)
	}

	opt := &blob.StorageOption{
		ProjectID:     bucket.Uint64ProjectID(),
		ClusterRegion: bucket.Regions[0],
		AWSRoleARN:    domain.BuildDataflowServiceInfraRoleARN(*bucket.CloudAccountID),
	}
	bucketURI := bucket.URI()
	s, blobErr := blob.NewStorage(ctx, bucketURI, opt)
	if blobErr != nil {
		logger.Error("create storage client error", zap.String("bucket uri", bucketURI), zap.Error(blobErr))
		return blobErr
	}

	if deleteErr := s.DeleteSubDir(ctx, "", nil); deleteErr != nil {
		logger.Error("delete bucket error", zap.Error(deleteErr))
		return deleteErr
	}
	return nil
}