package data_pipeline

import (
	"context"
	"fmt"
	"time"

	"github.com/go-co-op/gocron"
	data_pipeline_model "github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline"
	"github.com/tidbcloud/dataflow-service-ng/app/worker/common"
)

const workerName = "DataPipelineObservationWorker"

func NewWorkerScheduler(
	_ context.Context,
	loc *time.Location,
	intervalSeconds uint32,
	observation data_pipeline_model.ObservationModel,
) (*gocron.Scheduler, error) {
	if intervalSeconds == 0 {
		return nil, fmt.Errorf("data pipeline observation interval must be positive")
	}
	if observation == nil {
		return nil, fmt.Errorf("data pipeline observation model is required")
	}
	scheduler := gocron.NewScheduler(loc)
	scheduler.SingletonModeAll()
	if _, err := scheduler.Every(int(intervalSeconds)).Seconds().Do(
		common.WorkerFunWrapper(workerName, func(ctx context.Context) error {
			return observation.SyncObservations(ctx)
		}),
	); err != nil {
		return nil, err
	}
	return scheduler, nil
}