package worker

import (
	"context"
	"time"

	"github.com/go-co-op/gocron"
	"github.com/tidbcloud/dataflow-service/app/worker/backup"
	"github.com/tidbcloud/dataflow-service/app/worker/billing"
	"github.com/tidbcloud/dataflow-service/app/worker/biz_changefeed"
	"github.com/tidbcloud/dataflow-service/app/worker/bucket"
	"github.com/tidbcloud/dataflow-service/app/worker/private_link"
	"github.com/tidbcloud/dataflow-service/app/worker/replay"
	"github.com/tidbcloud/dataflow-service/app/worker/ticdc"
)

type Worker struct {
	schedulers []*gocron.Scheduler
}

func MustNewWorker(ctx context.Context, loc *time.Location) *Worker {
	bucketScheduler, err := bucket.NewWorkerScheduler(ctx, loc)
	if err != nil {
		panic(err)
	}

	backupScheduler, err := backup.NewWorkerScheduler(ctx, loc)
	if err != nil {
		panic(err)
	}

	billingScheduler, err := billing.NewWorkerScheduler(ctx, loc)
	if err != nil {
		panic(err)
	}
	bizChangefeedScheduler, err := biz_changefeed.NewWorkerScheduler(ctx, loc)
	if err != nil {
		panic(err)
	}
	privateLinkScheduler, err := private_link.NewWorkerScheduler(ctx, loc)
	if err != nil {
		panic(err)
	}
	ticdcScheduler, err := ticdc.NewWorkerScheduler(ctx, loc)
	if err != nil {
		panic(err)
	}

	replayScheduler, err := replay.NewWorkerScheduler(ctx, loc)
	if err != nil {
		panic(err)
	}

	w := &Worker{}
	w.schedulers = append(w.schedulers,
		bucketScheduler,
		backupScheduler,
		billingScheduler,
		bizChangefeedScheduler,
		privateLinkScheduler,
		ticdcScheduler,
		replayScheduler,
	)

	return w
}

func (w *Worker) StartWorker() {
	for _, scheduler := range w.schedulers {
		scheduler.StartAsync()
	}
}

func (w *Worker) StopWorker() {
	for _, scheduler := range w.schedulers {
		scheduler.Stop()
	}
}