package bucket

import (
	"context"
	"time"

	"github.com/go-co-op/gocron"
	"github.com/tidbcloud/dataflow-service/app/container"
	"github.com/tidbcloud/dataflow-service/app/worker/common"
	"github.com/tidbcloud/dataflow-service/internal/config"
)

func NewWorkerScheduler(ctx context.Context, loc *time.Location) (*gocron.Scheduler, error) {
	// always run in singleton mode
	s := gocron.NewScheduler(loc)
	s.SingletonModeAll()

	env := config.Global().Env
	switch env {
	case "nightly", "staging":
		// do the job every hour in dev and staging for testing and exposing bugs
		if _, err := s.Every(1).Hour().
			Do(common.WorkerFunWrapper("CleanGCPClusterBucket", CleanGCPClusterBucket)); err != nil {
			return nil, err
		}
		if _, err := s.Every(1).Hour().
			Do(common.WorkerFunWrapper("CleanGCPProjectBucket", CleanGCPProjectBucket)); err != nil {
			return nil, err
		}
		if _, err := s.Every(1).Hour().
			Do(common.WorkerFunWrapper("CleanAWSProjectBucket", CleanAWSProjectBucket)); err != nil {
			return nil, err
		}
		if _, err := s.Every(1).Hour().
			Do(common.WorkerFunWrapper("CleanAWSProjectBucketReplication",
				CleanAWSProjectBucketReplication)); err != nil {
			return nil, err
		}
	case "prod":
		// do the job every day
		if _, err := s.Every(1).Day().
			Do(common.WorkerFunWrapper("CleanGCPClusterBucket", CleanGCPClusterBucket)); err != nil {
			return nil, err
		}
		if _, err := s.Every(1).Day().
			Do(common.WorkerFunWrapper("CleanGCPProjectBucket", CleanGCPProjectBucket)); err != nil {
			return nil, err
		}
		if _, err := s.Every(1).Day().
			Do(common.WorkerFunWrapper("CleanAWSProjectBucket", CleanAWSProjectBucket)); err != nil {
			return nil, err
		}
		if _, err := s.Every(1).Day().
			Do(common.WorkerFunWrapper("CleanAWSProjectBucketReplication",
				CleanAWSProjectBucketReplication)); err != nil {
			return nil, err
		}
	}

	return s, nil
}

func CleanGCPClusterBucket(ctx context.Context) error {
	return container.Global().GCPClusterBucketWorkerV2.RecycleBucket(ctx)
}

func CleanGCPProjectBucket(ctx context.Context) error {
	return container.Global().GCPProjectBucketWorkerV2.RecycleBucket(ctx)
}

func CleanAWSProjectBucket(ctx context.Context) error {
	return container.Global().AWSBucketWorkerV2.RecycleBucket(ctx)
}

func CleanAWSProjectBucketReplication(ctx context.Context) error {
	return container.Global().AWSBucketReplicationWorker.RecycleBucketReplication(ctx)
}