agentic-coding-workflow/internal/jobs/worker_test.go

//go:build integration

package jobs_test

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

	"github.com/google/uuid"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"github.com/yan1h/agent-coding-workflow/internal/infra/clock"
	"github.com/yan1h/agent-coding-workflow/internal/jobs"
)

type recordingHandler struct {
	typ       jobs.JobType
	calls     atomic.Int32
	wantErr   error
	beforeRet func()
}

func (h *recordingHandler) Type() jobs.JobType { return h.typ }
func (h *recordingHandler) Handle(ctx context.Context, job jobs.Job) error {
	h.calls.Add(1)
	if h.beforeRet != nil {
		h.beforeRet()
	}
	return h.wantErr
}

func TestWorker_HandlesAndCompletes(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	repo, _ := setupRepo(t)
	_, err := repo.Enqueue(ctx, jobs.TypeWebhookDeliver, json.RawMessage(`{}`), time.Now(), 5)
	require.NoError(t, err)

	h := &recordingHandler{typ: jobs.TypeWebhookDeliver}
	w := jobs.NewWorker(repo, map[jobs.JobType]jobs.Handler{jobs.TypeWebhookDeliver: h},
		jobs.WorkerOptions{
			ID:           "test#1",
			PollInterval: 5 * time.Millisecond,
			Backoff:      []time.Duration{30 * time.Second},
			Clock:        clock.Real(),
			Log:          discardLogger(),
		})
	cctx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
	defer cancel()
	go w.Run(cctx)
	require.Eventually(t, func() bool { return h.calls.Load() >= 1 }, 200*time.Millisecond, 5*time.Millisecond)
	cancel()
	w.Wait()
}

func TestWorker_RetryableErrorReschedules(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	repo, _ := setupRepo(t)
	enqueued, err := repo.Enqueue(ctx, jobs.TypeWebhookDeliver, json.RawMessage(`{}`), time.Now(), 5)
	require.NoError(t, err)

	h := &recordingHandler{typ: jobs.TypeWebhookDeliver, wantErr: errors.New("transient")}
	w := jobs.NewWorker(repo, map[jobs.JobType]jobs.Handler{jobs.TypeWebhookDeliver: h},
		jobs.WorkerOptions{
			ID:           "t#1",
			PollInterval: 5 * time.Millisecond,
			Backoff:      []time.Duration{30 * time.Second, 2 * time.Minute},
			Clock:        clock.Real(),
			Log:          discardLogger(),
		})
	cctx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
	defer cancel()
	go w.Run(cctx)
	require.Eventually(t, func() bool { return h.calls.Load() >= 1 }, 200*time.Millisecond, 5*time.Millisecond)
	cancel()
	w.Wait()

	got, err := repo.Get(ctx, enqueued.ID)
	require.NoError(t, err)
	assert.Equal(t, jobs.StatusPending, got.Status)
	assert.Equal(t, int32(1), got.Attempts)
	require.NotNil(t, got.LastError)
	assert.Equal(t, "transient", *got.LastError)
}

func TestWorker_PermanentErrorDeadLetters(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	repo, _ := setupRepo(t)
	enqueued, err := repo.Enqueue(ctx, jobs.TypeWebhookDeliver, json.RawMessage(`{}`), time.Now(), 5)
	require.NoError(t, err)
	h := &recordingHandler{typ: jobs.TypeWebhookDeliver, wantErr: jobs.Permanent(errors.New("4xx config error"))}
	w := jobs.NewWorker(repo, map[jobs.JobType]jobs.Handler{jobs.TypeWebhookDeliver: h},
		jobs.WorkerOptions{ID: "t", PollInterval: 5 * time.Millisecond, Backoff: []time.Duration{time.Second}, Clock: clock.Real(), Log: discardLogger()})
	cctx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
	defer cancel()
	go w.Run(cctx)
	require.Eventually(t, func() bool {
		got, _ := repo.Get(ctx, enqueued.ID)
		return got != nil && got.Status == jobs.StatusDead
	}, 300*time.Millisecond, 10*time.Millisecond)
	cancel()
	w.Wait()
}

func TestWorker_MaxAttemptsExhaustedDeadLetters(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	repo, _ := setupRepo(t)
	enqueued, err := repo.Enqueue(ctx, jobs.TypeWebhookDeliver, json.RawMessage(`{}`), time.Now(), 2) // max=2
	require.NoError(t, err)
	h := &recordingHandler{typ: jobs.TypeWebhookDeliver, wantErr: errors.New("transient")}
	w := jobs.NewWorker(repo, map[jobs.JobType]jobs.Handler{jobs.TypeWebhookDeliver: h},
		jobs.WorkerOptions{ID: "t", PollInterval: 2 * time.Millisecond, Backoff: []time.Duration{time.Millisecond}, Clock: clock.Real(), Log: discardLogger()})
	cctx, cancel := context.WithTimeout(ctx, 500*time.Millisecond)
	defer cancel()
	go w.Run(cctx)
	require.Eventually(t, func() bool {
		got, _ := repo.Get(ctx, enqueued.ID)
		return got != nil && got.Status == jobs.StatusDead
	}, 500*time.Millisecond, 10*time.Millisecond)
	cancel()
	w.Wait()
}

func TestWorker_HandlerPanicTreatedAsRetry(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	repo, _ := setupRepo(t)
	enqueued, err := repo.Enqueue(ctx, jobs.TypeWebhookDeliver, json.RawMessage(`{}`), time.Now(), 5)
	require.NoError(t, err)
	h := &recordingHandler{typ: jobs.TypeWebhookDeliver, beforeRet: func() { panic("kaboom") }}
	w := jobs.NewWorker(repo, map[jobs.JobType]jobs.Handler{jobs.TypeWebhookDeliver: h},
		jobs.WorkerOptions{ID: "t", PollInterval: 5 * time.Millisecond, Backoff: []time.Duration{time.Second}, Clock: clock.Real(), Log: discardLogger()})
	cctx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
	defer cancel()
	go w.Run(cctx)
	require.Eventually(t, func() bool {
		got, _ := repo.Get(ctx, enqueued.ID)
		return got != nil && got.Status == jobs.StatusPending && got.Attempts >= 1
	}, 300*time.Millisecond, 10*time.Millisecond)
	cancel()
	w.Wait()
}

func TestWorker_UnknownTypeDeadLetters(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	repo, _ := setupRepo(t)
	enqueued, err := repo.Enqueue(ctx, "unknown.type", json.RawMessage(`{}`), time.Now(), 5)
	require.NoError(t, err)
	w := jobs.NewWorker(repo, map[jobs.JobType]jobs.Handler{},
		jobs.WorkerOptions{ID: "t", PollInterval: 5 * time.Millisecond, Backoff: []time.Duration{time.Second}, Clock: clock.Real(), Log: discardLogger()})
	cctx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
	defer cancel()
	go w.Run(cctx)
	require.Eventually(t, func() bool {
		got, _ := repo.Get(ctx, enqueued.ID)
		return got != nil && got.Status == jobs.StatusDead
	}, 200*time.Millisecond, 10*time.Millisecond)
	cancel()
	w.Wait()
}

func TestWorker_DeadLetterHookCalled(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	repo, _ := setupRepo(t)
	enqueued, err := repo.Enqueue(ctx, jobs.TypeWebhookDeliver, json.RawMessage(`{}`), time.Now(), 5)
	require.NoError(t, err)

	h := &recordingHandler{typ: jobs.TypeWebhookDeliver, wantErr: jobs.Permanent(errors.New("4xx"))}
	var called atomic.Int32
	var seenID atomic.Value // uuid.UUID
	w := jobs.NewWorker(repo, map[jobs.JobType]jobs.Handler{jobs.TypeWebhookDeliver: h},
		jobs.WorkerOptions{
			ID: "t", PollInterval: 5 * time.Millisecond, Backoff: []time.Duration{time.Second},
			Clock: clock.Real(), Log: discardLogger(),
			OnDeadLetter: func(_ context.Context, j jobs.Job, _ error) {
				seenID.Store(j.ID)
				called.Add(1)
			},
		})
	cctx, cancel := context.WithTimeout(ctx, 500*time.Millisecond)
	defer cancel()
	go w.Run(cctx)
	require.Eventually(t, func() bool { return called.Load() >= 1 }, 500*time.Millisecond, 10*time.Millisecond)
	cancel()
	w.Wait()
	if id, ok := seenID.Load().(uuid.UUID); ok {
		assert.Equal(t, enqueued.ID, id)
	}
}