summercms/modules/conga/worker_test.go

package conga

import (
	"context"
	"errors"
	"reflect"
	"strings"
	"testing"
	"time"

	"git.golem15.com/golem15/summercms/modules/backpack"
	"git.golem15.com/golem15/summercms/modules/compass"
	"git.golem15.com/golem15/summercms/modules/pact"
	"git.golem15.com/golem15/summercms/modules/party"
	"github.com/riverqueue/river/rivertype"
	"gorm.io/gorm"
)

type completeArgs struct {
	Mode string `json:"mode"`
}

func (completeArgs) Kind() string { return "conga_test_complete" }

// TestOutcomeComplete covers a job that completes its own row: status
// COMPLETE, progress at progress_max, and the River job completed. A job
// that returns nil without completing leaves the row as it is.
func TestOutcomeComplete(t *testing.T) {
	db, dsn := migratedDB(t)
	app, gdb := testApp(t, db, dsn, nil)
	m, err := From(app)
	if err != nil {
		t.Fatal(err)
	}
	job := Job(func(ctx context.Context, a completeArgs) error {
		id, _ := JobID(ctx)
		switch a.Mode {
		case "complete":
			if err := m.StartJob(ctx, id, 3); err != nil {
				return err
			}
			if err := m.UpdateJobState(ctx, id, 2, map[string]any{"step": 2}); err != nil {
				return err
			}
			return m.CompleteJob(ctx, id, map[string]any{"imported": 3})
		default:
			return nil
		}
	})
	startTestWorker(t, app, WorkerOptions{}, job)
	ctx := t.Context()

	id, err := m.Dispatch(ctx, gdb, completeArgs{Mode: "complete"}, DispatchOpts{Label: "Complete"})
	if err != nil {
		t.Fatal(err)
	}
	rec := waitStatus(t, m, id, StatusComplete)
	if rec.Progress != 3 || rec.ProgressMax != 3 {
		t.Fatalf("progress = %d/%d, want 3/3", rec.Progress, rec.ProgressMax)
	}
	if got := mustMetadata(t, m, id); !reflect.DeepEqual(got, map[string]any{"imported": float64(3)}) {
		t.Fatalf("metadata = %v", got)
	}
	waitRiverState(t, gdb, rec.RiverJobID, "completed")

	quiet, err := m.Dispatch(ctx, gdb, completeArgs{Mode: "quiet"}, DispatchOpts{Label: "Quiet"})
	if err != nil {
		t.Fatal(err)
	}
	qrec := mustGet(t, m, quiet)
	waitRiverState(t, gdb, qrec.RiverJobID, "completed")
	if qrec = mustGet(t, m, quiet); qrec.Status != StatusInProgress {
		t.Fatalf("status of a job that did not complete its row = %d, want IN_PROGRESS", qrec.Status)
	}
}

type testAppDB struct{ gdb *gorm.DB }

// outcomeEnv runs failArgs jobs that report the row status at the start of
// each attempt, then fail or panic, with immediate retries.
func outcomeEnv(t *testing.T) (*Manager, *testAppDB, chan Status) {
	t.Helper()
	db, dsn := migratedDB(t)
	app, gdb := testApp(t, db, dsn, nil)
	m, err := From(app)
	if err != nil {
		t.Fatal(err)
	}
	seen := make(chan Status, 16)
	job := Job(func(ctx context.Context, a failArgs) error {
		if id, ok := JobID(ctx); ok {
			if rec, err := m.Get(ctx, id); err == nil {
				seen <- rec.Status
			}
		}
		if a.Mode == "panic" {
			panic("boom")
		}
		return errors.New("import failed")
	})
	startTestWorker(t, app, WorkerOptions{pollInterval: 100 * time.Millisecond, retryPolicy: immediateRetry{}}, job)
	return m, &testAppDB{gdb: gdb}, seen
}

// TestOutcomeFailFinalAttemptOnly covers D-03: errors on attempts before
// the last keep the row IN_PROGRESS for River's retries; the final failed
// attempt records ERROR with the error text, and River discards the job.
func TestOutcomeFailFinalAttemptOnly(t *testing.T) {
	m, env, seen := outcomeEnv(t)
	ctx := t.Context()
	id, err := m.Dispatch(ctx, env.gdb, failArgs{Mode: "error"}, DispatchOpts{Label: "Retry", MaxAttempts: 3, Metadata: map[string]any{"file": "a.csv"}})
	if err != nil {
		t.Fatal(err)
	}
	rec := waitStatus(t, m, id, StatusError)
	for attempt := 1; attempt <= 3; attempt++ {
		select {
		case st := <-seen:
			if st != StatusInProgress {
				t.Fatalf("status at the start of attempt %d = %d, want %d", attempt, st, StatusInProgress)
			}
		case <-time.After(5 * time.Second):
			t.Fatalf("attempt %d did not run", attempt)
		}
	}
	if got := mustMetadata(t, m, id); !reflect.DeepEqual(got, map[string]any{"file": "a.csv", "error": "import failed"}) {
		t.Fatalf("metadata = %v, want the dispatch metadata plus the error text", got)
	}
	waitRiverState(t, env.gdb, rec.RiverJobID, "discarded")
	var attempts int
	if err := env.gdb.Raw(`SELECT attempt FROM river_job WHERE id = ?`, *rec.RiverJobID).Scan(&attempts).Error; err != nil || attempts != 3 {
		t.Fatalf("attempts = %d (err %v), want 3", attempts, err)
	}
}

// TestOutcomePanicBecomesError covers D-03: a panic on the final attempt is
// recovered and recorded like an error.
func TestOutcomePanicBecomesError(t *testing.T) {
	m, env, _ := outcomeEnv(t)
	id, err := m.Dispatch(t.Context(), env.gdb, failArgs{Mode: "panic"}, DispatchOpts{Label: "Panic", MaxAttempts: 1})
	if err != nil {
		t.Fatal(err)
	}
	waitStatus(t, m, id, StatusError)
	got, _ := mustMetadata(t, m, id)["error"].(string)
	if !strings.Contains(got, "panicked") || !strings.Contains(got, "boom") {
		t.Fatalf("metadata error = %q, want the panic text", got)
	}
}

// TestOutcomeSkipIsCompleteWithMetadata covers the WinterCMS skip rule:
// there is no skipped status; skipped work is COMPLETE with
// {"skipped": true}.
func TestOutcomeSkipIsCompleteWithMetadata(t *testing.T) {
	db, dsn := migratedDB(t)
	app, gdb := testApp(t, db, dsn, nil)
	m, err := From(app)
	if err != nil {
		t.Fatal(err)
	}
	job := Job(func(ctx context.Context, a completeArgs) error {
		id, _ := JobID(ctx)
		return m.CompleteJob(ctx, id, map[string]any{"skipped": true})
	})
	startTestWorker(t, app, WorkerOptions{}, job)
	id, err := m.Dispatch(t.Context(), gdb, completeArgs{Mode: "skip"}, DispatchOpts{Label: "Skip", Count: 2})
	if err != nil {
		t.Fatal(err)
	}
	rec := waitStatus(t, m, id, StatusComplete)
	if rec.Progress != 2 {
		t.Fatalf("progress = %d, want progress_max 2", rec.Progress)
	}
	if got := mustMetadata(t, m, id); !reflect.DeepEqual(got, map[string]any{"skipped": true}) {
		t.Fatalf("metadata = %v", got)
	}
}

// foreignJobsPlugin declares a job that conga did not build.
type foreignJobsPlugin struct{}

func (foreignJobsPlugin) ID() string                   { return "acme.foreign" }
func (foreignJobsPlugin) Requires() []string           { return nil }
func (foreignJobsPlugin) Register(*backpack.App) error { return nil }
func (foreignJobsPlugin) Boot(*backpack.App) error     { return nil }
func (foreignJobsPlugin) Jobs() []pact.Job             { return []pact.Job{otherJob{}} }

// TestStartWorkerRejectsNonCongaPluginJob covers the plugin job contract: a
// pact.HasJobs job not built by conga.Job fails the worker start naming
// the plugin.
func TestStartWorkerRejectsNonCongaPluginJob(t *testing.T) {
	_, err := StartWorker(t.Context(), backpack.New(nil), []party.Plugin{foreignJobsPlugin{}}, WorkerOptions{})
	if !errors.Is(err, ErrNotCongaJob) || !strings.Contains(err.Error(), "acme.foreign") {
		t.Fatalf("err = %v, want ErrNotCongaJob naming acme.foreign", err)
	}
}

// TestStartWorkerErrors covers the refusals of StartWorker and the job
// adapter.
func TestStartWorkerErrors(t *testing.T) {
	if _, err := StartWorker(t.Context(), backpack.New(nil), nil, WorkerOptions{}); !errors.Is(err, ErrNoDatabase) {
		t.Fatalf("no database: %v, want ErrNoDatabase", err)
	}
	if _, err := StartWorker(t.Context(), nil, nil, WorkerOptions{}); err == nil {
		t.Fatal("nil app accepted")
	}

	db, dsn := migratedDB(t)
	app, _ := testApp(t, db, dsn, nil)
	if _, err := StartWorker(t.Context(), app, nil, WorkerOptions{Queues: []string{"nope"}}); !errors.Is(err, ErrUnknownQueue) || !strings.Contains(err.Error(), "known queues: default, scheduled") {
		t.Fatalf("unknown queue: %v", err)
	}
	if _, err := StartWorker(t.Context(), app, []party.Plugin{&jobsPlugin{id: "acme.nil", jobs: []pact.Job{Job[pingArgs](nil)}}}, WorkerOptions{}); err == nil || !strings.Contains(err.Error(), "has no function") {
		t.Fatalf("job without a function: %v", err)
	}
	noDSN, _ := testApp(t, db, "", nil)
	if _, err := StartWorker(t.Context(), noDSN, nil, WorkerOptions{}); err == nil || !strings.Contains(err.Error(), "database.dsn is empty") {
		t.Fatalf("empty dsn: %v", err)
	}
	var nilWorker *Worker
	if nilWorker.Queues() != nil || nilWorker.Stop(context.Background()) != nil {
		t.Fatal("nil worker is not inert")
	}

	j := Job(func(context.Context, pingArgs) error { return errors.New("ran") })
	if err := j.Work(context.Background(), failArgs{}); err == nil || !strings.Contains(err.Error(), "got args of type") {
		t.Fatalf("wrong args type: %v", err)
	}
	if err := j.Work(context.Background(), pingArgs{}); err == nil || err.Error() != "ran" {
		t.Fatalf("Work = %v, want the job's own error", err)
	}
	if err := Job[pingArgs](nil).Work(context.Background(), pingArgs{}); err == nil {
		t.Fatal("job without a function ran")
	}
	if id, ok := summerJobID(&rivertype.JobRow{Metadata: []byte(`{"summer_job_id":"x"}`)}); ok || id != 0 {
		t.Fatal("malformed metadata gave a job id")
	}
	if _, ok := summerJobID(nil); ok {
		t.Fatal("nil row gave a job id")
	}
	if err := (idleWorker{}).Work(context.Background(), nil); err != nil || (idleArgs{}).Kind() != "summercms_conga_idle" {
		t.Fatal("idle worker is not a no-op")
	}
}

// TestWorkerStopFallsBackToHardStop covers Worker.Stop when its ctx ends
// before running jobs finish: the jobs are cancelled and the stop returns.
func TestWorkerStopFallsBackToHardStop(t *testing.T) {
	db, dsn := migratedDB(t)
	app, gdb := testApp(t, db, dsn, nil)
	m, err := From(app)
	if err != nil {
		t.Fatal(err)
	}
	started := make(chan struct{}, 1)
	job := Job(func(ctx context.Context, a blockArgs) error {
		started <- struct{}{}
		<-ctx.Done()
		return ctx.Err()
	}, Timeout(time.Minute))
	w, err := StartWorker(t.Context(), app, plugins(job), WorkerOptions{})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := m.Dispatch(t.Context(), gdb, blockArgs{}, DispatchOpts{Label: "Hard stop", MaxAttempts: 1}); err != nil {
		t.Fatal(err)
	}
	select {
	case <-started:
	case <-time.After(5 * time.Second):
		t.Fatal("job did not start")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
	defer cancel()
	done := make(chan error, 1)
	go func() { done <- w.Stop(ctx) }()
	select {
	case <-done:
	case <-time.After(10 * time.Second):
		t.Fatal("Stop did not fall back to a hard stop")
	}
	if err := m.Register(Job(func(context.Context, completeArgs) error { return nil })); err != nil {
		t.Fatalf("registration did not reopen after Stop: %v", err)
	}
}

// TestQueueSettings covers queue.* parsing: max_attempts, job_timeout as
// seconds or a duration string, work_in_serve and per-queue workers.
func TestQueueSettings(t *testing.T) {
	cfg := func(t *testing.T, kv map[string]any) *backpack.App {
		t.Helper()
		c, err := compass.Open(compass.Options{Dir: t.TempDir(), Env: "testing", Environ: []string{}})
		if err != nil {
			t.Fatal(err)
		}
		for k, v := range kv {
			if err := c.Set(k, v); err != nil {
				t.Fatal(err)
			}
		}
		return backpack.New(c)
	}
	def := settingsFromApp(nil)
	if def.maxAttempts != defaultMaxAttempts || def.jobTimeout != defaultJobTimeout || !def.workInServe || len(def.queues) != 0 {
		t.Fatalf("defaults = %+v", def)
	}
	cases := []struct {
		name string
		kv   map[string]any
		want func(s settings) bool
	}{
		{"timeout_seconds", map[string]any{"queue.job_timeout": 90}, func(s settings) bool { return s.jobTimeout == 90*time.Second }},
		{"timeout_duration", map[string]any{"queue.job_timeout": "2m30s"}, func(s settings) bool { return s.jobTimeout == 150*time.Second }},
		{"timeout_invalid_keeps_default", map[string]any{"queue.job_timeout": "soon"}, func(s settings) bool { return s.jobTimeout == defaultJobTimeout }},
		{"timeout_negative_keeps_default", map[string]any{"queue.job_timeout": "-5s"}, func(s settings) bool { return s.jobTimeout == defaultJobTimeout }},
		{"max_attempts", map[string]any{"queue.max_attempts": 7}, func(s settings) bool { return s.maxAttempts == 7 }},
		{"work_in_serve_false", map[string]any{"queue.work_in_serve": false}, func(s settings) bool { return !s.workInServe }},
		{"queues", map[string]any{"queue.queues": map[string]any{"imports": 2, "mail": 0, " ": 3}}, func(s settings) bool {
			return len(s.queues) == 2 && s.queues["imports"] == 2 && s.queues["mail"] == defaultMaxWorkers
		}},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			if s := settingsFromApp(cfg(t, c.kv)); !c.want(s) {
				t.Fatalf("settings = %+v", s)
			}
		})
	}
	known := knownQueues(settings{queues: map[string]int{"imports": 2}}, map[string]congaJob{
		"k": Job(func(context.Context, pingArgs) error { return nil }, OnQueue("mail")).(congaJob),
	})
	if !reflect.DeepEqual(known, map[string]int{"default": defaultMaxWorkers, "imports": 2, "mail": defaultMaxWorkers}) {
		t.Fatalf("knownQueues = %v", known)
	}
	sel, err := selectQueues(known, []string{" imports "})
	if err != nil || !reflect.DeepEqual(sel, map[string]int{"imports": 2}) {
		t.Fatalf("selectQueues = %v, %v", sel, err)
	}
	if all, err := selectQueues(known, nil); err != nil || len(all) != 3 {
		t.Fatalf("selectQueues(nil) = %v, %v", all, err)
	}
}