package data_pipeline

import (
	"context"
	"sync/atomic"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/pkgv2/bizerr"
)

func TestNewWorkerScheduler(t *testing.T) {
	model := new(observationModelStub)
	scheduler, err := NewWorkerScheduler(context.Background(), time.UTC, 1, model)
	require.NoError(t, err)
	require.Len(t, scheduler.Jobs(), 1)

	scheduler.StartAsync()
	require.Eventually(t, func() bool {
		return atomic.LoadInt32(&model.calls) > 0
	}, 3*time.Second, 50*time.Millisecond)
	scheduler.Stop()
}

func TestNewWorkerSchedulerRejectsInvalidConfiguration(t *testing.T) {
	_, err := NewWorkerScheduler(context.Background(), time.UTC, 0, new(observationModelStub))
	require.Error(t, err)

	_, err = NewWorkerScheduler(context.Background(), time.UTC, 30, nil)
	require.Error(t, err)
}

type observationModelStub struct {
	calls int32
}

func (m *observationModelStub) SyncObservations(context.Context) bizerr.BizError {
	atomic.AddInt32(&m.calls, 1)
	return nil
}