ITWorx-Pulse-Public/internal/alertworker/worker_test.go

package alertworker

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

	"github.com/itworx/pulse/internal/alert"
)

type fakeSource struct {
	rules []alert.Rule
}

func (s fakeSource) ListEnabled(_ context.Context, limit int) ([]alert.Rule, error) {
	if limit > len(s.rules) {
		limit = len(s.rules)
	}
	return s.rules[:limit], nil
}

type trackingEvaluator struct {
	active atomic.Int32
	max    atomic.Int32
	calls  atomic.Int32
	wait   time.Duration
	err    error
}

func (e *trackingEvaluator) Evaluate(ctx context.Context, _ alert.Rule) error {
	e.calls.Add(1)
	active := e.active.Add(1)
	for {
		current := e.max.Load()
		if active <= current || e.max.CompareAndSwap(current, active) {
			break
		}
	}
	defer e.active.Add(-1)
	if e.wait > 0 {
		timer := time.NewTimer(e.wait)
		defer timer.Stop()
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-timer.C:
		}
	}
	return e.err
}

func testRule(id string) alert.Rule {
	return alert.Rule{Document: alert.Document{ID: id, EvaluationIntervalSeconds: 30}}
}

func testWorker(t *testing.T, source RuleSource, store LeaseStore, evaluator Evaluator, owner string, now time.Time, maxConcurrent int) *Worker {
	t.Helper()
	worker, err := New(source, store, evaluator, Config{MaxConcurrent: maxConcurrent, MaxBatch: 100, AttemptTimeout: 100 * time.Millisecond, LeaseTTL: time.Second, Owner: owner, Now: func() time.Time { return now }})
	if err != nil {
		t.Fatal(err)
	}
	return &worker
}

func TestDuplicateWorkersDoNotEvaluateSameSlotTwice(t *testing.T) {
	source := fakeSource{rules: []alert.Rule{testRule("rule-1")}}
	store := NewMemoryLeaseStore()
	evaluator := &trackingEvaluator{wait: 20 * time.Millisecond}
	now := time.Date(2026, 8, 2, 5, 0, 10, 0, time.UTC)
	first := testWorker(t, source, store, evaluator, "worker-1", now, 1)
	second := testWorker(t, source, store, evaluator, "worker-2", now, 1)
	var reports [2]RunReport
	var wait sync.WaitGroup
	wait.Add(2)
	go func() { defer wait.Done(); reports[0], _ = first.RunOnce(context.Background()) }()
	go func() { defer wait.Done(); reports[1], _ = second.RunOnce(context.Background()) }()
	wait.Wait()
	if evaluator.calls.Load() != 1 {
		t.Fatalf("evaluation calls = %d, want 1", evaluator.calls.Load())
	}
	if reports[0].Skipped+reports[1].Skipped != 1 {
		t.Fatalf("skips = %d, want 1", reports[0].Skipped+reports[1].Skipped)
	}
	third, err := first.RunOnce(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if third.Skipped != 1 || evaluator.calls.Load() != 1 {
		t.Fatalf("repeat report=%#v calls=%d", third, evaluator.calls.Load())
	}
}

func TestTimeoutIsVisibleInJobResult(t *testing.T) {
	source := fakeSource{rules: []alert.Rule{testRule("rule-timeout")}}
	store := NewMemoryLeaseStore()
	evaluator := &trackingEvaluator{wait: time.Second}
	now := time.Date(2026, 8, 2, 5, 0, 10, 0, time.UTC)
	worker, err := New(source, store, evaluator, Config{MaxConcurrent: 1, MaxBatch: 1, AttemptTimeout: 10 * time.Millisecond, LeaseTTL: time.Second, Owner: "worker-timeout", Now: func() time.Time { return now }})
	if err != nil {
		t.Fatal(err)
	}
	report, err := worker.RunOnce(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if report.Failed != 1 || len(report.Jobs) != 1 || report.Jobs[0].ErrorCode != "timeout" {
		t.Fatalf("timeout report=%#v", report)
	}
}

func TestShutdownCancelsOutstandingEvaluationSafely(t *testing.T) {
	source := fakeSource{rules: []alert.Rule{testRule("rule-shutdown")}}
	store := NewMemoryLeaseStore()
	started := make(chan struct{})
	evaluator := EvaluateFunc(func(ctx context.Context, _ alert.Rule) error {
		close(started)
		<-ctx.Done()
		return ctx.Err()
	})
	now := time.Date(2026, 8, 2, 5, 0, 10, 0, time.UTC)
	worker := testWorker(t, source, store, evaluator, "worker-shutdown", now, 1)
	ctx, cancel := context.WithCancel(context.Background())
	result := make(chan RunReport, 1)
	go func() {
		report, _ := worker.RunOnce(ctx)
		result <- report
	}()
	<-started
	cancel()
	report := <-result
	if report.Canceled != 1 || len(report.Jobs) != 1 || report.Jobs[0].ErrorCode != "shutdown" {
		t.Fatalf("shutdown report=%#v", report)
	}
}

func TestLoadIsBoundedByMaxConcurrentAndBatch(t *testing.T) {
	rules := make([]alert.Rule, 20)
	for i := range rules {
		rules[i] = testRule(alert.NewID())
	}
	store := NewMemoryLeaseStore()
	evaluator := &trackingEvaluator{wait: 2 * time.Millisecond}
	now := time.Date(2026, 8, 2, 5, 0, 10, 0, time.UTC)
	worker := testWorker(t, fakeSource{rules: rules}, store, evaluator, "worker-scale", now, 3)
	report, err := worker.RunOnce(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if report.Scheduled != 20 || report.Started != 20 || report.Completed != 20 || evaluator.max.Load() > 3 {
		t.Fatalf("bounded report=%#v max=%d", report, evaluator.max.Load())
	}
	if metrics := worker.Metrics(); metrics.JobsStarted != 20 || metrics.JobsCompleted != 20 {
		t.Fatalf("metrics=%#v", metrics)
	}
}

func TestInvalidWorkerConfigurationAndEvaluationError(t *testing.T) {
	_, err := New(fakeSource{}, NewMemoryLeaseStore(), EvaluateFunc(func(context.Context, alert.Rule) error { return nil }), Config{MaxConcurrent: 0, MaxBatch: 1, AttemptTimeout: time.Second, LeaseTTL: time.Second, Owner: "x", Now: time.Now})
	if !errors.Is(err, ErrInvalidConfig) {
		t.Fatalf("config error=%v", err)
	}
	source := fakeSource{rules: []alert.Rule{testRule("rule-error")}}
	store := NewMemoryLeaseStore()
	worker := testWorker(t, source, store, EvaluateFunc(func(context.Context, alert.Rule) error { return errors.New("upstream failed") }), "worker-error", time.Now().UTC(), 1)
	report, err := worker.RunOnce(context.Background())
	if err != nil || report.Failed != 1 || report.Jobs[0].ErrorCode != "evaluation_failed" {
		t.Fatalf("error report=%#v err=%v", report, err)
	}
}

func TestFailedSlotIsVisibleAndNotRetried(t *testing.T) {
	source := fakeSource{rules: []alert.Rule{testRule("rule-failed")}}
	store := NewMemoryLeaseStore()
	evaluator := &trackingEvaluator{err: errors.New("upstream failed")}
	now := time.Date(2026, 8, 2, 5, 0, 10, 0, time.UTC)
	worker := testWorker(t, source, store, evaluator, "worker-failed", now, 1)
	first, err := worker.RunOnce(context.Background())
	if err != nil || first.Failed != 1 || first.Jobs[0].ErrorCode != "evaluation_failed" {
		t.Fatalf("first report=%#v err=%v", first, err)
	}
	second, err := worker.RunOnce(context.Background())
	if err != nil || second.Skipped != 1 || evaluator.calls.Load() != 1 {
		t.Fatalf("second report=%#v calls=%d err=%v", second, evaluator.calls.Load(), err)
	}
}