Wire plan_ready, list_ready, new_suggestions and Send to Google Chat

Connect the features to the notification outbox so real events notify,
each stored by the new notify.EmitTx in the transaction of the change it
is about (Wake after commit):

- plan_ready:{weekStart} (/plan) when a draft job stores the AI's draft;
  a redraft shares the key and a manual swap never emits. The title reads
  "Next week's plan is ready" for the coming week.
- list_ready:{weekStart} (/plan/shopping) when a week's Supply run is
  first created, by generate or after Push to Calendar.
- new_suggestions:{batchId} (/discover) when a scheduled or manual batch
  ends done with cards; replenish batches never notify.
- plan.Service.EmitDayOfReminder(date) emits day_of_reminder:{date} for
  the Feature 20 scheduler.

Add POST /api/v1/shopping/{weekStart}/send-chat: shopping.SendToChat posts
the dinners with needed items and a link through chat.PostSupplyRun,
records sent_to_chat_at, and is idempotent per week unless
{"resend": true}. The Supply run's Send to Google Chat button calls it
("Send again" once sent) and shows failures in an ErrorBanner; the unused
Unavailable placeholder is removed.

Task: f0cd9d50-9c11-41be-a945-7bae9f5f28a4

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

internal/notify/notify.go

@@ -6,7 +6,8 @@
 // row in notification_outbox keyed by (event key, channel). The event key is
 // the caller's idempotency key, so emitting the same event twice stores
 // nothing the second time and never touches a row already stored, whatever
-// its status.
+// its status. EmitTx does the same inside the caller's transaction, so an
+// event is stored with the change it is about; the caller then calls Wake.
 //
 // The worker started by Start delivers due rows through the Sender
 // registered for their channel. A delivered row is marked sent. A failed
@@ -142,6 +143,20 @@ func NewService(st *store.Store, logger *slog.Logger) (*Service, error) {
 // row with ev.Key is left as it is. An invalid event is ErrInvalidEvent and
 // stores nothing; a type with no preference row is an error.
 func (s *Service) Emit(ctx context.Context, ev Event) error {
+	err := s.st.WithTx(ctx, func(tx *sql.Tx) error {
+		return s.EmitTx(ctx, tx, ev)
+	})
+	if err != nil {
+		return err
+	}
+	s.Wake()
+	return nil
+}
+
+// EmitTx stores ev's notifications as Emit does, in tx, so they are
+// committed with the change the event is about or not at all. The caller
+// calls Wake once tx has committed.
+func (s *Service) EmitTx(ctx context.Context, tx *sql.Tx, ev Event) error {
 	if err := validate(ev); err != nil {
 		return err
 	}
@@ -154,42 +169,33 @@ func (s *Service) Emit(ctx context.Context, ev Event) error {
 		return fmt.Errorf("notify: encode payload of %s: %w", ev.Key, err)
 	}
 
-	var queued []Channel
-	err = s.st.WithTx(ctx, func(tx *sql.Tx) error {
-		q := db.New(tx)
-		pref, err := q.GetNotificationPref(ctx, string(ev.Type))
-		if errors.Is(err, sql.ErrNoRows) {
-			return fmt.Errorf("notify: no notification preference for %s", ev.Type)
-		}
-		if err != nil {
-			return fmt.Errorf("notify: read notification preference for %s: %w", ev.Type, err)
-		}
-		now := timestamp(s.now())
-		for _, ch := range enabledChannels(pref) {
-			n, err := q.InsertNotification(ctx, db.InsertNotificationParams{
-				ID:            store.NewID(),
-				EventKey:      ev.Key,
-				Channel:       string(ch),
-				Payload:       string(data),
-				NextAttemptAt: sql.NullString{String: now, Valid: true},
-				Now:           now,
-			})
-			if err != nil {
-				return fmt.Errorf("notify: store %s notification %s: %w", ch, ev.Key, err)
-			}
-			if n == 1 {
-				queued = append(queued, ch)
-			}
-		}
-		return nil
-	})
+	q := db.New(tx)
+	pref, err := q.GetNotificationPref(ctx, string(ev.Type))
+	if errors.Is(err, sql.ErrNoRows) {
+		return fmt.Errorf("notify: no notification preference for %s", ev.Type)
+	}
 	if err != nil {
-		return err
+		return fmt.Errorf("notify: read notification preference for %s: %w", ev.Type, err)
 	}
-	s.logger.InfoContext(ctx, "notification emitted", "event_key", ev.Key, "queued", queued)
-	if len(queued) > 0 {
-		s.wakeWorker()
+	now := timestamp(s.now())
+	var queued []Channel
+	for _, ch := range enabledChannels(pref) {
+		n, err := q.InsertNotification(ctx, db.InsertNotificationParams{
+			ID:            store.NewID(),
+			EventKey:      ev.Key,
+			Channel:       string(ch),
+			Payload:       string(data),
+			NextAttemptAt: sql.NullString{String: now, Valid: true},
+			Now:           now,
+		})
+		if err != nil {
+			return fmt.Errorf("notify: store %s notification %s: %w", ch, ev.Key, err)
+		}
+		if n == 1 {
+			queued = append(queued, ch)
+		}
 	}
+	s.logger.InfoContext(ctx, "notification stored in transaction", "event_key", ev.Key, "queued", queued)
 	return nil
 }
 

internal/notify/notify_test.go

@@ -753,3 +753,34 @@ func TestWorkerDeliversEmittedEvent(t *testing.T) {
 		t.Fatalf("%d deliveries, want 1", n)
 	}
 }
+
+// TestEmitTxCommitsWithTheCallersTransaction: EmitTx's rows are stored only
+// when the caller's transaction commits.
+func TestEmitTxCommitsWithTheCallersTransaction(t *testing.T) {
+	ctx := context.Background()
+	svc, st, _ := newTestService(t)
+	setPref(t, st, PlanReady, true, false)
+	const key = "plan_ready:2026-10-12"
+	rollback := errors.New("roll back")
+	err := st.WithTx(ctx, func(tx *sql.Tx) error {
+		if err := svc.EmitTx(ctx, tx, planReady(key)); err != nil {
+			return err
+		}
+		return rollback
+	})
+	if !errors.Is(err, rollback) {
+		t.Fatalf("WithTx = %v", err)
+	}
+	if rows := outbox(t, st, key); len(rows) != 0 {
+		t.Fatalf("rows after rollback = %v, want none", rows)
+	}
+	if err := st.WithTx(ctx, func(tx *sql.Tx) error { return svc.EmitTx(ctx, tx, planReady(key)) }); err != nil {
+		t.Fatal(err)
+	}
+	if r := onlyRow(t, st, key, ChannelPush); r.Status != "pending" {
+		t.Errorf("row after commit = %+v, want pending", r)
+	}
+	if err := st.WithTx(ctx, func(tx *sql.Tx) error { return svc.EmitTx(ctx, tx, Event{Type: PlanReady, Key: key}) }); !errors.Is(err, ErrInvalidEvent) {
+		t.Errorf("EmitTx of an invalid event = %v, want ErrInvalidEvent", err)
+	}
+}

internal/notify/notifytest/notifytest.go

@@ -0,0 +1,65 @@
+// Package notifytest builds notify services for tests of the features that
+// emit notifications, and reads back what they stored.
+package notifytest
+
+import (
+	"encoding/json"
+	"io"
+	"log/slog"
+	"testing"
+
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/household"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/notify"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store"
+)
+
+// New returns a notify service over st, with st's notification preferences
+// seeded, so every event type goes to push and plan_ready and list_ready to
+// Google Chat too. Its worker is not started: what is emitted stays pending.
+func New(t *testing.T, st *store.Store) *notify.Service {
+	t.Helper()
+	if err := household.SeedNotificationPrefs(t.Context(), st); err != nil {
+		t.Fatal(err)
+	}
+	svc, err := notify.NewService(st, slog.New(slog.NewTextHandler(io.Discard, nil)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return svc
+}
+
+// Row is one stored notification.
+type Row struct {
+	Key     string
+	Channel notify.Channel
+	Payload notify.Payload
+}
+
+// Rows returns every notification stored in st whose event is of type ev,
+// oldest first.
+func Rows(t *testing.T, st *store.Store, ev notify.EventType) []Row {
+	t.Helper()
+	rs, err := st.Reader.QueryContext(t.Context(),
+		`SELECT event_key, channel, payload FROM notification_outbox WHERE event_key LIKE ? ORDER BY created_at, channel`,
+		string(ev)+":%")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rs.Close()
+	var out []Row
+	for rs.Next() {
+		var r Row
+		var payload string
+		if err := rs.Scan(&r.Key, &r.Channel, &payload); err != nil {
+			t.Fatal(err)
+		}
+		if err := json.Unmarshal([]byte(payload), &r.Payload); err != nil {
+			t.Fatal(err)
+		}
+		out = append(out, r)
+	}
+	if err := rs.Err(); err != nil {
+		t.Fatal(err)
+	}
+	return out
+}

internal/notify/worker.go

@@ -104,8 +104,9 @@ func (s *Service) Wait() {
 	s.wg.Wait()
 }
 
-// wakeWorker gives the worker a token to look for due rows.
-func (s *Service) wakeWorker() {
+// Wake tells the worker to look for due rows, as after EmitTx's
+// transaction commits.
+func (s *Service) Wake() {
 	select {
 	case s.wake <- struct{}{}:
 	default: // the worker already has a token and will look again