Require an explicit payload on POST /api/v1/jobs

An omitted payload was silently replaced with {}, contrary to the
fail-fast and no-silent-fallback rules. It is now refused with 400
invalid_request "payload: required"; a null payload stays refused.
Kinds that take no input, such as noop, are sent {}.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/jobs/jobs.go

@@ -68,6 +68,14 @@ var codePattern = regexp.MustCompile(`^[a-z][a-z0-9_]{0,63}$`)
 // ErrNotFound is returned by Get for an unknown job id.
 var ErrNotFound = errors.New("jobs: job not found")
 
+// ErrUnknownKind is returned by Runner.Enqueue for a kind with no
+// registered handler.
+var ErrUnknownKind = errors.New("jobs: unknown job kind")
+
+// ErrInvalidPayload is returned by Runner.Enqueue for a payload that does
+// not decode into the kind's payload type.
+var ErrInvalidPayload = errors.New("jobs: payload does not decode")
+
 // Failure is a handler error reporting a failure code the household may
 // see. Code must be one the job's kind declared at registration; the runner
 // stores it as the job's error. Any other handler error, including a Failure
@@ -106,6 +114,13 @@ type Job struct {
 // never quotes an error.
 type runFunc func(ctx context.Context, payload string, publish func(json.RawMessage) error) (result json.RawMessage, code, reason string)
 
+// handler is a registered kind: run runs a job, and check reports whether
+// a payload decodes into the kind's payload type.
+type handler struct {
+	run   runFunc
+	check func(payload []byte) bool
+}
+
 // Runner owns the job handlers and the worker pool.
 type Runner struct {
 	st      *store.Store
@@ -118,7 +133,7 @@ type Runner struct {
 	claim func(ctx context.Context, now sql.NullString) (db.Job, error)
 
 	mu       sync.Mutex
-	handlers map[string]runFunc
+	handlers map[string]handler
 	started  bool
 
 	wg sync.WaitGroup
@@ -141,7 +156,7 @@ func NewRunner(st *store.Store, logger *slog.Logger, workers int) (*Runner, erro
 		workers:  workers,
 		wake:     make(chan struct{}, workers),
 		claim:    db.New(st.Writer).ClaimNextJob,
-		handlers: make(map[string]runFunc),
+		handlers: make(map[string]handler),
 	}, nil
 }
 
@@ -179,11 +194,9 @@ func Register[P any, R Public](r *Runner, kind string, h Handler[P, R], failureC
 	if _, ok := r.handlers[kind]; ok {
 		return nil, fmt.Errorf("jobs: register %q: kind already registered", kind)
 	}
-	r.handlers[kind] = func(ctx context.Context, payload string, publish func(json.RawMessage) error) (json.RawMessage, string, string) {
-		var p P
-		dec := json.NewDecoder(bytes.NewReader([]byte(payload)))
-		dec.DisallowUnknownFields()
-		if err := dec.Decode(&p); err != nil {
+	run := func(ctx context.Context, payload string, publish func(json.RawMessage) error) (json.RawMessage, string, string) {
+		p, err := decodePayload[P]([]byte(payload))
+		if err != nil {
 			return nil, CodeInternal, "payload does not decode"
 		}
 		result, err := h(ctx, p, func(partial R) error {
@@ -209,9 +222,23 @@ func Register[P any, R Public](r *Runner, kind string, h Handler[P, R], failureC
 		}
 		return data, "", ""
 	}
+	check := func(payload []byte) bool {
+		_, err := decodePayload[P](payload)
+		return err == nil
+	}
+	r.handlers[kind] = handler{run: run, check: check}
 	return &Kind[P]{r: r, name: kind}, nil
 }
 
+// decodePayload decodes payload into P, refusing unknown fields.
+func decodePayload[P any](payload []byte) (P, error) {
+	var p P
+	dec := json.NewDecoder(bytes.NewReader(payload))
+	dec.DisallowUnknownFields()
+	err := dec.Decode(&p)
+	return p, err
+}
+
 // publicJSON returns v's PublicJSON, refusing anything that is not one JSON
 // value. Its error never quotes PublicJSON's error.
 func publicJSON(v Public) (json.RawMessage, error) {
@@ -229,19 +256,43 @@ func (k *Kind[P]) Enqueue(ctx context.Context, payload P) (string, error) {
 	if err != nil {
 		return "", fmt.Errorf("jobs: enqueue %s: encode payload: %w", k.name, err)
 	}
+	return k.r.enqueue(ctx, k.name, data)
+}
+
+// Enqueue stores a queued job of kind with its JSON payload and returns its
+// id, as Kind.Enqueue does for a kind known at compile time. It returns
+// ErrUnknownKind for a kind with no registered handler and
+// ErrInvalidPayload for a payload that does not decode into the kind's
+// payload type; neither stores a job.
+func (r *Runner) Enqueue(ctx context.Context, kind string, payload json.RawMessage) (string, error) {
+	r.mu.Lock()
+	h, ok := r.handlers[kind]
+	r.mu.Unlock()
+	if !ok {
+		return "", ErrUnknownKind
+	}
+	if !h.check(payload) {
+		return "", ErrInvalidPayload
+	}
+	return r.enqueue(ctx, kind, payload)
+}
+
+// enqueue stores a queued job of kind with its encoded payload and wakes a
+// worker.
+func (r *Runner) enqueue(ctx context.Context, kind string, payload []byte) (string, error) {
 	id := store.NewID()
 	now := store.Now().String()
-	if err := db.New(k.r.st.Writer).CreateJob(ctx, db.CreateJobParams{
+	if err := db.New(r.st.Writer).CreateJob(ctx, db.CreateJobParams{
 		ID:        id,
-		Kind:      k.name,
-		Payload:   string(data),
+		Kind:      kind,
+		Payload:   string(payload),
 		CreatedAt: now,
 		UpdatedAt: now,
 	}); err != nil {
-		return "", fmt.Errorf("jobs: enqueue %s: %w", k.name, err)
+		return "", fmt.Errorf("jobs: enqueue %s: %w", kind, err)
 	}
 	select {
-	case k.r.wake <- struct{}{}:
+	case r.wake <- struct{}{}:
 	default: // every worker already has a token and will look again
 	}
 	return id, nil
@@ -365,7 +416,7 @@ func (r *Runner) run(ctx context.Context, job db.Job) {
 		return nil
 	}
 
-	result, code, reason := h(ctx, job.Payload, publish)
+	result, code, reason := h.run(ctx, job.Payload, publish)
 	if code != "" {
 		if ctx.Err() != nil {
 			logger.WarnContext(write, "job interrupted by shutdown; it fails at the next start")

internal/jobs/jobs_test.go

@@ -568,3 +568,39 @@ func TestClaimErrorIsRetried(t *testing.T) {
 		t.Fatalf("claim error text logged: %s", out)
 	}
 }
+
+func TestRunnerEnqueueByKindName(t *testing.T) {
+	t.Parallel()
+	st := migratedStore(t)
+	r := newRunner(t, st, 1)
+	register(t, r, "count", func(_ context.Context, p countPayload, _ func(countResult) error) (countResult, error) {
+		return countResult{Done: p.Steps}, nil
+	})
+	ctx := context.Background()
+
+	if _, err := r.Enqueue(ctx, "no_such_kind", json.RawMessage(`{}`)); !errors.Is(err, ErrUnknownKind) {
+		t.Errorf("unknown kind error = %v, want ErrUnknownKind", err)
+	}
+	for _, payload := range []string{`{"steps":"three"}`, `{"other":1}`, `[]`, ``} {
+		if _, err := r.Enqueue(ctx, "count", json.RawMessage(payload)); !errors.Is(err, ErrInvalidPayload) {
+			t.Errorf("payload %q error = %v, want ErrInvalidPayload", payload, err)
+		}
+	}
+	var n int
+	if err := st.Reader.QueryRow("SELECT COUNT(*) FROM jobs").Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	if n != 0 {
+		t.Errorf("refused enqueues stored %d jobs, want 0", n)
+	}
+
+	id, err := r.Enqueue(ctx, "count", json.RawMessage(`{"steps":3}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	start(t, r)
+	job := waitFor(t, r, id, finished)
+	if job.Kind != "count" || job.Status != StatusSucceeded || string(job.Result) != `{"done":3}` {
+		t.Errorf("job = %+v (result %s), want succeeded with {\"done\":3}", job, job.Result)
+	}
+}

internal/server/jobs.go

@@ -1,7 +1,11 @@
 package server
 
 import (
+	"bytes"
+	"encoding/json"
 	"errors"
+	"fmt"
+	"io"
 	"log/slog"
 	"net/http"
 
@@ -10,6 +14,80 @@ import (
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
 )
 
+// maxJobBody bounds a POST /jobs request body.
+const maxJobBody = 64 << 10
+
+// jobRequest is the POST /jobs body. Payload is required; a job kind that
+// takes no input is sent the empty object {}.
+type jobRequest struct {
+	Kind    string          `json:"kind"`
+	Payload json.RawMessage `json:"payload"`
+}
+
+// jobCreated is the POST /jobs response.
+type jobCreated struct {
+	ID string `json:"id"`
+}
+
+// postJob serves POST /api/v1/jobs: it enqueues a job of a registered kind
+// and returns 202 with its id, which GET /api/v1/jobs/{id} reports on.
+func postJob(logger *slog.Logger, runner *jobs.Runner) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		req, err := decodeJobRequest(http.MaxBytesReader(w, r.Body, maxJobBody))
+		if err != nil {
+			writeError(w, r, logger, http.StatusBadRequest, "invalid_request", err.Error())
+			return
+		}
+		id, err := runner.Enqueue(r.Context(), req.Kind, req.Payload)
+		switch {
+		case errors.Is(err, jobs.ErrUnknownKind):
+			writeError(w, r, logger, http.StatusBadRequest, "unknown_kind", "unknown job kind")
+			return
+		case errors.Is(err, jobs.ErrInvalidPayload):
+			writeError(w, r, logger, http.StatusBadRequest, "invalid_payload", "payload does not match the job kind")
+			return
+		case err != nil:
+			logger.ErrorContext(r.Context(), "enqueue job", "kind", req.Kind, "error", err)
+			writeError(w, r, logger, http.StatusInternalServerError, "internal", "internal error")
+			return
+		}
+		writeJSON(w, r, logger, http.StatusAccepted, jobCreated{ID: id})
+	}
+}
+
+// decodeJobRequest reads a single JSON object with a non-empty kind and a
+// payload, refusing unknown fields, an omitted or null payload and trailing
+// data.
+func decodeJobRequest(body io.Reader) (jobRequest, error) {
+	data, err := io.ReadAll(body)
+	if err != nil {
+		var tooLarge *http.MaxBytesError
+		if errors.As(err, &tooLarge) {
+			return jobRequest{}, fmt.Errorf("request body exceeds %d bytes", tooLarge.Limit)
+		}
+		return jobRequest{}, errors.New("cannot read request body")
+	}
+	var req jobRequest
+	dec := json.NewDecoder(bytes.NewReader(data))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&req); err != nil {
+		return jobRequest{}, errors.New("request body must be a JSON object with kind and payload")
+	}
+	if _, err := dec.Token(); err != io.EOF {
+		return jobRequest{}, errors.New("request body must be a single JSON object")
+	}
+	if req.Kind == "" {
+		return jobRequest{}, errors.New("kind: required")
+	}
+	if req.Payload == nil {
+		return jobRequest{}, errors.New("payload: required")
+	}
+	if bytes.Equal(req.Payload, []byte("null")) {
+		return jobRequest{}, errors.New("payload: must not be null")
+	}
+	return req, nil
+}
+
 // getJob serves GET /api/v1/jobs/{id}: the job's id, kind, status, latest
 // result and error.
 func getJob(logger *slog.Logger, runner *jobs.Runner) http.HandlerFunc {

internal/server/jobs_test.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"slices"
+	"strings"
 	"testing"
 	"time"
 
@@ -151,3 +152,95 @@ func TestGetJobRequiresSession(t *testing.T) {
 		t.Errorf("status = %d, want 401", rec.Code)
 	}
 }
+
+// TestPostJobThenGetProgressAndResult drives the whole flow over the
+// router: POST enqueues a progressive "draft" job, GET reads its partial
+// result while the handler is held mid-run, then its final result.
+func TestPostJobThenGetProgressAndResult(t *testing.T) {
+	t.Parallel()
+	release := make(chan struct{})
+	runner, _ := startedRunner(t, release, nil)
+	router := newTestRouterWith(t, "dev", newTestSettings(t), runner)
+
+	rec, body := serve(t, router, signedIn(http.MethodPost, "/api/v1/jobs", strings.NewReader(`{"kind":"draft","payload":{}}`)))
+	if rec.Code != http.StatusAccepted {
+		t.Fatalf("status = %d, want 202: %s", rec.Code, rec.Body)
+	}
+	id, _ := body["id"].(string)
+	if keys := slices.Sorted(maps.Keys(body)); !slices.Equal(keys, []string{"id"}) || id == "" {
+		t.Fatalf("body = %v, want only a job id", body)
+	}
+
+	mid := pollJob(t, router, id, func(b map[string]any) bool { return resultDays(b) == 1 })
+	if mid["id"] != id || mid["kind"] != "draft" || mid["status"] != "running" || mid["error"] != nil {
+		t.Errorf("mid-run body = %v, want running draft job %s with no error", mid, id)
+	}
+
+	close(release)
+	done := pollJob(t, router, id, func(b map[string]any) bool { return b["status"] == "succeeded" })
+	if done["id"] != id || resultDays(done) != 7 || done["error"] != nil {
+		t.Errorf("finished body = %v, want result {\"days\":7} and no error", done)
+	}
+}
+
+func TestPostJobRefusesBadRequests(t *testing.T) {
+	t.Parallel()
+	runner, _ := startedRunner(t, nil, nil)
+	router := newTestRouterWith(t, "dev", newTestSettings(t), runner)
+
+	for _, tc := range []struct {
+		name, body, code, message string
+	}{
+		{"unknown kind", `{"kind":"no_such_kind","payload":{}}`, "unknown_kind", "unknown job kind"},
+		{"payload for another type", `{"kind":"draft","payload":{"days":3}}`, "invalid_payload", "payload does not match the job kind"},
+		{"payload not an object", `{"kind":"draft","payload":[1]}`, "invalid_payload", "payload does not match the job kind"},
+		{"missing payload", `{"kind":"draft"}`, "invalid_request", "payload: required"},
+		{"null payload", `{"kind":"draft","payload":null}`, "invalid_request", "payload: must not be null"},
+		{"missing kind", `{"payload":{}}`, "invalid_request", "kind: required"},
+		{"empty body", ``, "invalid_request", "request body must be a JSON object with kind and payload"},
+		{"not an object", `["draft"]`, "invalid_request", "request body must be a JSON object with kind and payload"},
+		{"unknown field", `{"kind":"draft","payload":{},"extra":1}`, "invalid_request", "request body must be a JSON object with kind and payload"},
+		{"trailing data", `{"kind":"draft","payload":{}} {}`, "invalid_request", "request body must be a single JSON object"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			rec := httptest.NewRecorder()
+			router.ServeHTTP(rec, signedIn(http.MethodPost, "/api/v1/jobs", strings.NewReader(tc.body)))
+			if rec.Code != http.StatusBadRequest {
+				t.Fatalf("status = %d, want 400: %s", rec.Code, rec.Body)
+			}
+			var body errorBody
+			if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
+				t.Fatal(err)
+			}
+			if want := (errorBody{Error: apiError{Code: tc.code, Message: tc.message}}); body != want {
+				t.Errorf("body = %+v, want %+v", body, want)
+			}
+		})
+	}
+}
+
+func TestPostJobNeedsSessionAndCSRF(t *testing.T) {
+	t.Parallel()
+	router := newTestRouter(t, "dev")
+	const body = `{"kind":"noop","payload":{}}`
+
+	rec := httptest.NewRecorder()
+	router.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/api/v1/jobs", strings.NewReader(body)))
+	if rec.Code != http.StatusUnauthorized || rec.Body.String() != unauthenticatedJSON {
+		t.Errorf("no session: %d %s, want 401 %s", rec.Code, rec.Body, unauthenticatedJSON)
+	}
+
+	for _, header := range []string{"", "wrong-token"} {
+		req := httptest.NewRequest(http.MethodPost, "/api/v1/jobs", strings.NewReader(body))
+		req.AddCookie(validSessionCookie)
+		if header != "" {
+			req.Header.Set(CSRFHeader, header)
+		}
+		rec := httptest.NewRecorder()
+		router.ServeHTTP(rec, req)
+		if rec.Code != http.StatusForbidden || rec.Body.String() != csrfJSON {
+			t.Errorf("CSRF header %q: %d %s, want 403 %s", header, rec.Code, rec.Body, csrfJSON)
+		}
+	}
+}

internal/server/server.go

@@ -30,7 +30,7 @@ type RouteMounter interface {
 // only from a peer in trusted (see resolveOrigin). version is reported by
 // GET /health; auth mounts the /auth routes; sessions guards /api/v1 and
 // /media; settings serves /api/v1/settings/household; runner serves
-// /api/v1/jobs/{id}; assets is the web build
+// POST /api/v1/jobs and GET /api/v1/jobs/{id}; assets is the web build
 // served for every other GET. /health, /auth and the web build are public.
 // Requests to /api/v1 other than GET also need the session's CSRF token in CSRFHeader.
 func NewRouter(logger *slog.Logger, version string, trusted forwarded.Trusted, auth RouteMounter, sessions SessionLookup, settings *household.SettingsService, runner *jobs.Runner, assets fs.FS) (http.Handler, error) {
@@ -68,6 +68,7 @@ func NewRouter(logger *slog.Logger, version string, trusted forwarded.Trusted, a
 	apiV1.Get("/me", me(logger))
 	apiV1.Get("/settings/household", getHouseholdSettings(logger, settings))
 	apiV1.Patch("/settings/household", patchHouseholdSettings(logger, settings))
+	apiV1.Post("/jobs", postJob(logger, runner))
 	apiV1.Get("/jobs/{id}", getJob(logger, runner))
 	r.Route("/api", func(r chi.Router) {
 		r.MethodNotAllowed(methodNotAllowed)