package worker import ( "context" "errors" "io" "log/slog" "strings" "sync" "testing" "time" "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]) } } func TestRunnerRetriesClaimErrors(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() store := &fakeJobStore{ claimErrors: []error{errors.New("relation \"background_jobs\" does not exist")}, job: &jobs.Job{ ID: "job-4", Kind: jobs.KindBootstrapStructureMaterialize, }, cancel: cancel, } runner := NewRunner(store, slog.New(slog.NewTextHandler(io.Discard, nil)), time.Millisecond) handlerCalled := false runner.Register(jobs.KindBootstrapStructureMaterialize, func(ctx context.Context, job jobs.Job) error { handlerCalled = true 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 after claim retry") } if store.claimAttempts < 2 { t.Fatalf("expected at least two claim attempts, got %d", store.claimAttempts) } if len(store.succeeded) != 1 || store.succeeded[0] != "job-4" { 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) } } type fakeJobStore struct { mu sync.Mutex job *jobs.Job claimed bool claimErrors []error claimAttempts int 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() store.claimAttempts++ if len(store.claimErrors) > 0 { err := store.claimErrors[0] store.claimErrors = store.claimErrors[1:] return nil, err } 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 }