package worker

import (
	"context"
	"strings"
	"time"

	"github.com/go-co-op/gocron"
	"github.com/tidbcloud/dataflow-service-ng/app/container"
	data_pipeline_model "github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline"
	"github.com/tidbcloud/dataflow-service-ng/app/worker/billing"
	"github.com/tidbcloud/dataflow-service-ng/app/worker/biz_changefeed"
	data_pipeline_worker "github.com/tidbcloud/dataflow-service-ng/app/worker/data_pipeline"
	import_into_worker "github.com/tidbcloud/dataflow-service-ng/app/worker/import_into"
	premium_dm_worker "github.com/tidbcloud/dataflow-service-ng/app/worker/premium_dm"
	"github.com/tidbcloud/dataflow-service-ng/app/worker/sink_endpoint"
	"github.com/tidbcloud/dataflow-service-ng/app/worker/ticdc"
	service_config "github.com/tidbcloud/dataflow-service-ng/internal/config"
)

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

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

	bizChangefeedScheduler, err := biz_changefeed.NewWorkerScheduler(ctx, loc)
	if err != nil {
		panic(err)
	}
	ticdcScheduler, err := ticdc.NewWorkerScheduler(ctx, loc)
	if err != nil {
		panic(err)
	}
	privateLinkScheduler, err := sink_endpoint.NewWorkerScheduler(ctx, loc)
	if err != nil {
		panic(err)
	}
	premiumDMScheduler, err := premium_dm_worker.NewWorkerScheduler(ctx, loc)
	if err != nil {
		panic(err)
	}
	importIntoScheduler, err := import_into_worker.NewWorkerScheduler(ctx, loc)
	if err != nil {
		panic(err)
	}
	dataPipelineScheduler, err := newDataPipelineWorkerScheduler(
		ctx,
		loc,
		container.Global().DataPipelineConfig,
		container.Global().DataPipelineObservationModel,
	)
	if err != nil {
		panic(err)
	}

	w := &Worker{}
	w.schedulers = append(w.schedulers,
		billingScheduler,
		bizChangefeedScheduler,
		ticdcScheduler,
		privateLinkScheduler,
		premiumDMScheduler,
		importIntoScheduler,
	)
	if dataPipelineScheduler != nil {
		w.schedulers = append(w.schedulers, dataPipelineScheduler)
	}

	return w
}

func newDataPipelineWorkerScheduler(
	ctx context.Context,
	loc *time.Location,
	cfg service_config.DataPipelineConfig,
	model data_pipeline_model.ObservationModel,
) (*gocron.Scheduler, error) {
	// The worker only has something to observe once a destination is configured.
	if strings.TrimSpace(cfg.LakeBaseURL) == "" {
		return nil, nil
	}
	return data_pipeline_worker.NewWorkerScheduler(
		ctx,
		loc,
		cfg.ObservationIntervalSeconds,
		model,
	)
}

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

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