Refactor: improve code quality and worker flow

2026-06-26 18:32:49 +01:00
parent adcc9afe05
commit 7e62ff6d9a
16 changed files with 1444 additions and 358 deletions
@@ -0,0 +1,118 @@
+package worker
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"strings"
+	"time"
+
+	"moku-backend/internal/jobs"
+)
+
+type JobStore interface {
+	ClaimNext(ctx context.Context) (*jobs.Job, error)
+	MarkSucceeded(ctx context.Context, jobID string) error
+	MarkFailed(ctx context.Context, jobID, failure string) error
+}
+
+type Handler func(ctx context.Context, job jobs.Job) error
+
+type Runner struct {
+	store        JobStore
+	logger       *slog.Logger
+	pollInterval time.Duration
+	handlers     map[string]Handler
+}
+
+func NewRunner(store JobStore, logger *slog.Logger, pollInterval time.Duration) *Runner {
+	interval := pollInterval
+	if interval <= 0 {
+		interval = time.Second
+	}
+
+	return &Runner{
+		store:        store,
+		logger:       logger,
+		pollInterval: interval,
+		handlers:     make(map[string]Handler),
+	}
+}
+
+func (runner *Runner) Register(kind string, handler Handler) {
+	runner.handlers[strings.TrimSpace(kind)] = handler
+}
+
+func (runner *Runner) Run(ctx context.Context) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		default:
+		}
+
+		job, err := runner.store.ClaimNext(ctx)
+		if err != nil {
+			if ctx.Err() != nil {
+				return nil
+			}
+
+			return err
+		}
+
+		if job == nil {
+			if err := waitForNextPoll(ctx, runner.pollInterval); err != nil {
+				return nil
+			}
+
+			continue
+		}
+
+		handler, ok := runner.handlers[job.Kind]
+		if !ok {
+			failure := fmt.Sprintf("no handler registered for job kind %q", job.Kind)
+			if err := runner.store.MarkFailed(ctx, job.ID, failure); err != nil {
+				return err
+			}
+
+			runner.logger.Error("worker job failed", "jobID", job.ID, "kind", job.Kind, "error", failure)
+			continue
+		}
+
+		if err := handler(ctx, *job); err != nil {
+			if ctx.Err() != nil {
+				return nil
+			}
+
+			failure := strings.TrimSpace(err.Error())
+			if failure == "" {
+				failure = "job handler returned an empty error"
+			}
+
+			if markErr := runner.store.MarkFailed(ctx, job.ID, failure); markErr != nil {
+				return markErr
+			}
+
+			runner.logger.Error("worker job failed", "jobID", job.ID, "kind", job.Kind, "error", failure)
+			continue
+		}
+
+		if err := runner.store.MarkSucceeded(ctx, job.ID); err != nil {
+			return err
+		}
+
+		runner.logger.Info("worker job succeeded", "jobID", job.ID, "kind", job.Kind)
+	}
+}
+
+func waitForNextPoll(ctx context.Context, interval time.Duration) error {
+	timer := time.NewTimer(interval)
+	defer timer.Stop()
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
+	}
+}
@@ -0,0 +1,161 @@
+package worker
+
+import (
+	"context"
+	"errors"
+	"io"
+	"log/slog"
+	"strings"
+	"sync"
+	"testing"
+
+	"moku-backend/internal/jobs"
+)
+
+func TestRunnerProcessesRegisteredJob(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	store := &fakeJobStore{
+		job: &jobs.Job{
+			ID:      "job-1",
+			Kind:    jobs.KindBootstrapStructureMaterialize,
+			Payload: []byte(`{"installationId":"installation-1"}`),
+		},
+		cancel: cancel,
+	}
+	runner := NewRunner(store, slog.New(slog.NewTextHandler(io.Discard, nil)), 0)
+
+	handlerCalled := false
+	runner.Register(jobs.KindBootstrapStructureMaterialize, func(ctx context.Context, job jobs.Job) error {
+		handlerCalled = true
+		if job.ID != "job-1" {
+			t.Fatalf("expected job id job-1, got %s", job.ID)
+		}
+		return nil
+	})
+
+	if err := runner.Run(ctx); err != nil {
+		t.Fatalf("runner returned error: %v", err)
+	}
+
+	if !handlerCalled {
+		t.Fatal("expected handler to be called")
+	}
+	if len(store.succeeded) != 1 || store.succeeded[0] != "job-1" {
+		t.Fatalf("expected job to be marked succeeded once, got %#v", store.succeeded)
+	}
+	if len(store.failed) != 0 {
+		t.Fatalf("expected no failed jobs, got %#v", store.failed)
+	}
+}
+
+func TestRunnerMarksFailedWhenHandlerErrors(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	store := &fakeJobStore{
+		job: &jobs.Job{
+			ID:   "job-2",
+			Kind: jobs.KindBootstrapStructureMaterialize,
+		},
+		cancel: cancel,
+	}
+	runner := NewRunner(store, slog.New(slog.NewTextHandler(io.Discard, nil)), 0)
+	runner.Register(jobs.KindBootstrapStructureMaterialize, func(ctx context.Context, job jobs.Job) error {
+		return errors.New("boom")
+	})
+
+	if err := runner.Run(ctx); err != nil {
+		t.Fatalf("runner returned error: %v", err)
+	}
+
+	if len(store.succeeded) != 0 {
+		t.Fatalf("expected no succeeded jobs, got %#v", store.succeeded)
+	}
+	if len(store.failed) != 1 {
+		t.Fatalf("expected one failed job, got %#v", store.failed)
+	}
+	if store.failed[0].jobID != "job-2" {
+		t.Fatalf("expected failed job id job-2, got %#v", store.failed[0])
+	}
+	if !strings.Contains(store.failed[0].failure, "boom") {
+		t.Fatalf("expected failure to mention handler error, got %#v", store.failed[0])
+	}
+}
+
+func TestRunnerMarksFailedWhenHandlerMissing(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	store := &fakeJobStore{
+		job: &jobs.Job{
+			ID:   "job-3",
+			Kind: "unknown.kind",
+		},
+		cancel: cancel,
+	}
+	runner := NewRunner(store, slog.New(slog.NewTextHandler(io.Discard, nil)), 0)
+
+	if err := runner.Run(ctx); err != nil {
+		t.Fatalf("runner returned error: %v", err)
+	}
+
+	if len(store.failed) != 1 {
+		t.Fatalf("expected one failed job, got %#v", store.failed)
+	}
+	if !strings.Contains(store.failed[0].failure, "no handler registered") {
+		t.Fatalf("expected missing handler failure, got %#v", store.failed[0])
+	}
+}
+
+type fakeJobStore struct {
+	mu        sync.Mutex
+	job       *jobs.Job
+	claimed   bool
+	succeeded []string
+	failed    []fakeFailure
+	cancel    context.CancelFunc
+}
+
+type fakeFailure struct {
+	jobID   string
+	failure string
+}
+
+func (store *fakeJobStore) ClaimNext(ctx context.Context) (*jobs.Job, error) {
+	store.mu.Lock()
+	defer store.mu.Unlock()
+
+	if store.claimed || store.job == nil {
+		return nil, nil
+	}
+
+	store.claimed = true
+	job := *store.job
+	return &job, nil
+}
+
+func (store *fakeJobStore) MarkSucceeded(ctx context.Context, jobID string) error {
+	store.mu.Lock()
+	store.succeeded = append(store.succeeded, jobID)
+	store.mu.Unlock()
+
+	if store.cancel != nil {
+		store.cancel()
+	}
+
+	return nil
+}
+
+func (store *fakeJobStore) MarkFailed(ctx context.Context, jobID, failure string) error {
+	store.mu.Lock()
+	store.failed = append(store.failed, fakeFailure{jobID: jobID, failure: failure})
+	store.mu.Unlock()
+
+	if store.cancel != nil {
+		store.cancel()
+	}
+
+	return nil
+}