diff --git a/internal/jobs/jobs.go b/internal/jobs/jobs.go index 929d7f3..337addc 100644 --- a/internal/jobs/jobs.go +++ b/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") diff --git a/internal/jobs/jobs_test.go b/internal/jobs/jobs_test.go index 714aac4..1ab0822 100644 --- a/internal/jobs/jobs_test.go +++ b/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) + } +} diff --git a/internal/server/jobs.go b/internal/server/jobs.go index 2fdd15f..9f33d16 100644 --- a/internal/server/jobs.go +++ b/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 { diff --git a/internal/server/jobs_test.go b/internal/server/jobs_test.go index 6898242..7e8005b 100644 --- a/internal/server/jobs_test.go +++ b/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) + } + } +} diff --git a/internal/server/server.go b/internal/server/server.go index a01adf3..b1f3049 100644 --- a/internal/server/server.go +++ b/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)