Feature 13 integration fix: deck error surfacing, active-batch tracking, modal key scope, three-card deck (#111)

Co-authored-by: orchestrator-agent <20+orchestrator-agent@code.cubecraftcreations.com>

internal/server/suggestions.go

@@ -10,10 +10,6 @@ import (
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/suggest"
 )
 
-type deck struct {
-	Suggestions []suggest.Suggestion `json:"suggestions"`
-}
-
 // postSuggestionBatch starts a manual suggestion batch and answers 202 with
 // its batch_id and the job_id that GET /jobs/{id} reports on, without
 // waiting for the job. When a batch is already pending or running it
@@ -30,15 +26,15 @@ func postSuggestionBatch(logger *slog.Logger, svc *suggest.Service) http.Handler
 }
 
 // getSuggestionDeck answers GET /suggestions/deck with the pending
-// suggestions, oldest first.
+// suggestions, oldest first, and the batch pending or running, or null.
 func getSuggestionDeck(logger *slog.Logger, svc *suggest.Service) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
-		list, err := svc.Deck(r.Context())
+		d, err := svc.Deck(r.Context())
 		if err != nil {
 			writeSuggestionError(w, r, logger, "list suggestion deck", err)
 			return
 		}
-		writeJSON(w, r, logger, http.StatusOK, deck{Suggestions: list})
+		writeJSON(w, r, logger, http.StatusOK, d)
 	}
 }
 

internal/server/suggestions_test.go

@@ -148,8 +148,44 @@ func TestSuggestionDecisions(t *testing.T) {
 	if rec.Code != http.StatusOK {
 		t.Fatalf("decline = %d: %s", rec.Code, rec.Body)
 	}
-	if id, _ := body["replenish_job_id"].(string); id == "" {
-		t.Errorf("decline leaving 4 = %v, want a replenish_job_id", body)
+	replenish, _ := body["replenish_job_id"].(string)
+	if replenish == "" {
+		t.Fatalf("decline leaving 4 = %v, want a replenish_job_id", body)
+	}
+
+	// A fourth decision joins the batch already pending and reports its job.
+	rec, body = callDinners(t, h, http.MethodPost, suggestionsPath+"/"+ids[3]+"/accept", "")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("accept = %d: %s", rec.Code, rec.Body)
+	}
+	if body["replenish_job_id"] != replenish {
+		t.Errorf("accept leaving 3 = %v, want replenish_job_id %s", body, replenish)
+	}
+}
+
+func TestSuggestionDeckActiveBatch(t *testing.T) {
+	t.Parallel()
+	h, _, _ := newSuggestionsRouter(t, "Pho")
+
+	rec, body := callDinners(t, h, http.MethodGet, suggestionsPath+"/deck", "")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("GET deck = %d: %s", rec.Code, rec.Body)
+	}
+	if active, ok := body["active_batch"]; !ok || active != nil {
+		t.Fatalf("deck = %v, want active_batch null", body)
+	}
+
+	rec, started := callDinners(t, h, http.MethodPost, suggestionsPath+"/batches", "")
+	if rec.Code != http.StatusAccepted {
+		t.Fatalf("POST batches = %d: %s", rec.Code, rec.Body)
+	}
+	rec, body = callDinners(t, h, http.MethodGet, suggestionsPath+"/deck", "")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("GET deck = %d: %s", rec.Code, rec.Body)
+	}
+	active, _ := body["active_batch"].(map[string]any)
+	if active == nil || active["batch_id"] != started["batch_id"] || active["job_id"] != started["job_id"] || active["status"] != "pending" {
+		t.Errorf("deck active_batch = %v, want the started batch %v", body["active_batch"], started)
 	}
 }
 

internal/suggest/suggest.go

@@ -171,34 +171,47 @@ func (s *Service) StartBatch(ctx context.Context, trigger string) (Batch, error)
 	return b, nil
 }
 
-// startBatch does StartBatch's work in tx: the batch and the job that runs
-// it are committed together, so a batch is never seen without its job. A
-// batch still pending or running whose job is no longer queued or running
-// is failed first, so a batch whose job ended without recording its end
-// (the process stopped, or the write failed) never blocks a later one. The
-// caller wakes the job runner once tx has committed and Created is true.
-func (s *Service) startBatch(ctx context.Context, tx *sql.Tx, trigger string) (Batch, error) {
-	q := db.New(tx)
+// activeBatch returns the batch pending or running in q's transaction, or
+// nil when there is none. A batch still pending or running whose job is no
+// longer queued or running is failed first, so a batch whose job ended
+// without recording its end (the process stopped, or the write failed) is
+// never reported as active and never blocks a later one.
+func (s *Service) activeBatch(ctx context.Context, q *db.Queries) (*Batch, error) {
 	n, err := q.FailOrphanedSuggestionBatches(ctx, db.FailOrphanedSuggestionBatchesParams{
 		UpdatedAt: store.Now().String(),
 		Kind:      KindSuggestionBatch,
 	})
 	if err != nil {
-		return Batch{}, fmt.Errorf("suggest: fail orphaned batches: %w", err)
+		return nil, fmt.Errorf("suggest: fail orphaned batches: %w", err)
 	}
 	if n > 0 {
 		s.logger.WarnContext(ctx, "failed suggestion batches whose job ended", "count", n)
 	}
 	cur, err := q.GetActiveSuggestionBatch(ctx)
-	if err == nil {
-		jobID, err := q.GetSuggestionBatchJobID(ctx, db.GetSuggestionBatchJobIDParams{Kind: KindSuggestionBatch, BatchID: cur.ID})
-		if err != nil {
-			return Batch{}, fmt.Errorf("suggest: read job of batch %s: %w", cur.ID, err)
-		}
-		return Batch{ID: cur.ID, JobID: jobID, Trigger: cur.Trigger, Status: cur.Status}, nil
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil, nil
 	}
-	if !errors.Is(err, sql.ErrNoRows) {
-		return Batch{}, fmt.Errorf("suggest: read active batch: %w", err)
+	if err != nil {
+		return nil, fmt.Errorf("suggest: read active batch: %w", err)
+	}
+	jobID, err := q.GetSuggestionBatchJobID(ctx, db.GetSuggestionBatchJobIDParams{Kind: KindSuggestionBatch, BatchID: cur.ID})
+	if err != nil {
+		return nil, fmt.Errorf("suggest: read job of batch %s: %w", cur.ID, err)
+	}
+	return &Batch{ID: cur.ID, JobID: jobID, Trigger: cur.Trigger, Status: cur.Status}, nil
+}
+
+// startBatch does StartBatch's work in tx: the batch and the job that runs
+// it are committed together, so a batch is never seen without its job. The
+// caller wakes the job runner once tx has committed and Created is true.
+func (s *Service) startBatch(ctx context.Context, tx *sql.Tx, trigger string) (Batch, error) {
+	q := db.New(tx)
+	cur, err := s.activeBatch(ctx, q)
+	if err != nil {
+		return Batch{}, err
+	}
+	if cur != nil {
+		return *cur, nil
 	}
 	now := store.Now().String()
 	row, err := q.CreateSuggestionBatch(ctx, db.CreateSuggestionBatchParams{
@@ -261,25 +274,44 @@ type Suggestion struct {
 	CreatedAt       string   `json:"created_at"`
 }
 
-// Deck returns the pending suggestions whose dinner is still suggested,
-// oldest first.
-func (s *Service) Deck(ctx context.Context) ([]Suggestion, error) {
-	rows, err := db.New(s.st.Reader).ListDeck(ctx)
-	if err != nil {
-		return nil, fmt.Errorf("suggest: list deck: %w", err)
-	}
-	out := make([]Suggestion, len(rows))
-	for i, row := range rows {
-		out[i], err = fromRow(row.Suggestion, row.Dinner)
+// Deck is the household's deck: the pending suggestions whose dinner is
+// still suggested, oldest first, and the batch pending or running, if any,
+// so a client can follow the job producing more.
+type Deck struct {
+	Suggestions []Suggestion `json:"suggestions"`
+	ActiveBatch *Batch       `json:"active_batch"`
+}
+
+// Deck reads the deck and its active batch in one transaction, so both come
+// from the same snapshot. It is a write transaction because reading the
+// active batch first fails any batch whose job has ended (see activeBatch).
+func (s *Service) Deck(ctx context.Context) (Deck, error) {
+	var deck Deck
+	err := s.st.WithTx(ctx, func(tx *sql.Tx) error {
+		q := db.New(tx)
+		rows, err := q.ListDeck(ctx)
 		if err != nil {
-			return nil, err
+			return fmt.Errorf("suggest: list deck: %w", err)
 		}
+		deck.Suggestions = make([]Suggestion, len(rows))
+		for i, row := range rows {
+			deck.Suggestions[i], err = fromRow(row.Suggestion, row.Dinner)
+			if err != nil {
+				return err
+			}
+		}
+		deck.ActiveBatch, err = s.activeBatch(ctx, q)
+		return err
+	})
+	if err != nil {
+		return Deck{}, err
 	}
-	return out, nil
+	return deck, nil
 }
 
 // Decision is a suggestion once decided, with the id of the replenish
-// batch's job when the decision started one.
+// batch's job when the decision called for one: the batch it started, or
+// the one already pending or running that it joined.
 type Decision struct {
 	Suggestion     Suggestion `json:"suggestion"`
 	ReplenishJobID *string    `json:"replenish_job_id"`
@@ -303,8 +335,8 @@ func (s *Service) Decline(ctx context.Context, id, updatedBy string) (Decision,
 // dinnerStatus. Repeating a decision changes nothing and returns the
 // suggestion as it is; the opposite decision is a *DecidedError. A
 // decision that leaves the deck under ReplenishBelow starts a replenish
-// batch unless one is already pending or running, in the decision's own
-// transaction: the decision is never recorded without the replenish it
+// batch, or joins the one already pending or running, in the decision's own
+// transaction, and returns its job id either way: the decision is never recorded without the replenish it
 // calls for, and a failure to start the batch records neither.
 func (s *Service) decide(ctx context.Context, id, decision, dinnerStatus, updatedBy string) (Decision, error) {
 	if updatedBy == "" {
@@ -369,8 +401,7 @@ func (s *Service) decide(ctx context.Context, id, decision, dinnerStatus, update
 	if err != nil {
 		return Decision{}, err
 	}
-	created := replenish != nil && replenish.Created
-	if created {
+	if replenish != nil && replenish.Created {
 		s.batchJob.Wake()
 	}
 	sug, err := fromRow(row.Suggestion, row.Dinner)
@@ -378,7 +409,7 @@ func (s *Service) decide(ctx context.Context, id, decision, dinnerStatus, update
 		return Decision{}, err
 	}
 	d := Decision{Suggestion: sug}
-	if created {
+	if replenish != nil {
 		d.ReplenishJobID = &replenish.JobID
 	}
 	return d, nil

internal/suggest/suggest_test.go

@@ -2,6 +2,7 @@ package suggest_test
 
 import (
 	"context"
+	"database/sql"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -329,7 +330,7 @@ func (e *env) deckNames(t *testing.T) []string {
 		t.Fatal(err)
 	}
 	names := []string{}
-	for _, s := range deck {
+	for _, s := range deck.Suggestions {
 		names = append(names, s.Name)
 	}
 	return names
@@ -362,15 +363,15 @@ func TestBatchStoresGroundedSideQuests(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(deck) != 2 {
-		t.Fatalf("deck has %d suggestions, want 2", len(deck))
+	if len(deck.Suggestions) != 2 {
+		t.Fatalf("deck has %d suggestions, want 2", len(deck.Suggestions))
 	}
 	want := []struct{ name, reason, regular string }{
 		{"Chicken Shawarma Bowls", "Spiced chicken, like your Chicken Fajitas.", "Chicken Fajitas"},
 		{"Chili Verde", "Your Beef Chili gone green.", "Beef Chili"},
 	}
 	for i, w := range want {
-		s := deck[i]
+		s := deck.Suggestions[i]
 		if s.ID != res.SuggestionIDs[i] || s.Name != w.name || s.Reason != w.reason || s.Status != suggest.StatusPending {
 			t.Errorf("deck[%d] = %+v, want %s pending with reason %q", i, s, w.name, w.reason)
 		}
@@ -452,6 +453,69 @@ func TestDeckIsOldestFirst(t *testing.T) {
 	}
 }
 
+func TestDeckReportsTheActiveBatch(t *testing.T) {
+	e := newEnv(t) // runner not started: the batch stays pending
+	e.seedDeck(t, "Pho")
+	deck, err := e.svc.Deck(t.Context())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if deck.ActiveBatch != nil {
+		t.Fatalf("active batch = %+v with none pending or running, want nil", deck.ActiveBatch)
+	}
+	b, err := e.svc.StartBatch(t.Context(), suggest.TriggerManual)
+	if err != nil {
+		t.Fatal(err)
+	}
+	deck, err = e.svc.Deck(t.Context())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(deck.Suggestions) != 1 {
+		t.Errorf("deck has %d suggestions, want 1", len(deck.Suggestions))
+	}
+	want := suggest.Batch{ID: b.ID, JobID: b.JobID, Trigger: suggest.TriggerManual, Status: suggest.BatchPending}
+	if deck.ActiveBatch == nil || *deck.ActiveBatch != want {
+		t.Errorf("active batch = %+v, want %+v", deck.ActiveBatch, want)
+	}
+}
+
+func TestDeckFailsBatchInterruptedByRestart(t *testing.T) {
+	e := newEnv(t)
+	e.seedDeck(t, "Pho")
+	b, err := e.svc.StartBatch(t.Context(), suggest.TriggerManual)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// A previous process claimed the job and started the batch, then
+	// stopped before recording its end.
+	now := store.Now().String()
+	if _, err := db.New(e.st.Writer).ClaimNextJob(t.Context(), sql.NullString{String: now, Valid: true}); err != nil {
+		t.Fatal(err)
+	}
+	if n, err := db.New(e.st.Writer).StartSuggestionBatch(t.Context(), db.StartSuggestionBatchParams{UpdatedAt: now, ID: b.ID}); err != nil || n != 1 {
+		t.Fatalf("start batch: %d rows, %v", n, err)
+	}
+	e.start(t) // fails the running job as interrupted
+	if job := e.waitJob(t, b.JobID); job.Status != jobs.StatusFailed {
+		t.Fatalf("job status = %s, want failed", job.Status)
+	}
+
+	deck, err := e.svc.Deck(t.Context())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if deck.ActiveBatch != nil {
+		t.Errorf("active batch = %+v after its job failed, want nil", deck.ActiveBatch)
+	}
+	if len(deck.Suggestions) != 1 {
+		t.Errorf("deck has %d suggestions, want 1", len(deck.Suggestions))
+	}
+	if got := e.batchStatus(t, b.ID); got != suggest.BatchFailed {
+		t.Errorf("interrupted batch status = %s, want failed", got)
+	}
+}
+
 func TestDeckLeavesOutDinnersNoLongerSuggested(t *testing.T) {
 	e := newEnv(t)
 	ids := e.seedDeck(t, "Pho", "Ramen")
@@ -459,7 +523,7 @@ func TestDeckLeavesOutDinnersNoLongerSuggested(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if _, err := db.New(e.st.Writer).ArchiveDinner(t.Context(), db.ArchiveDinnerParams{ID: sug[0].DinnerID, UpdatedAt: store.Now().String()}); err != nil {
+	if _, err := db.New(e.st.Writer).ArchiveDinner(t.Context(), db.ArchiveDinnerParams{ID: sug.Suggestions[0].DinnerID, UpdatedAt: store.Now().String()}); err != nil {
 		t.Fatal(err)
 	}
 	if got := e.deckNames(t); !slices.Equal(got, []string{"Ramen"}) {
@@ -608,8 +672,10 @@ func TestReplenishIsSingleFlight(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if second.ReplenishJobID != nil {
-		t.Errorf("second decision enqueued replenish job %s while one is pending", *second.ReplenishJobID)
+	// The second decision joins the pending batch and reports its job, so a
+	// client that did not start it can follow it.
+	if second.ReplenishJobID == nil || *second.ReplenishJobID != *first.ReplenishJobID {
+		t.Errorf("second decision replenish job = %v, want the pending batch's job %s", second.ReplenishJobID, *first.ReplenishJobID)
 	}
 	manual, err := e.svc.StartBatch(t.Context(), suggest.TriggerManual)
 	if err != nil {