notify package and durable outbox worker (Feature 19) (#161)

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

cmd/server/main.go

@@ -24,6 +24,7 @@ import (
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/gcal"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/household"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/notify"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/plan"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/push"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/recipes"
@@ -144,6 +145,11 @@ func run(cfg config.Config, logger *slog.Logger) error {
 		return err
 	}
 
+	notifier, err := notify.NewService(db, logger)
+	if err != nil {
+		return err
+	}
+
 	sessions, err := auth.NewSessionManager(db, cfg.SessionIdle, cfg.SessionMax)
 	if err != nil {
 		return err
@@ -166,14 +172,18 @@ func run(cfg config.Config, logger *slog.Logger) error {
 	}
 
 	// Start fails the jobs a previous process left running before any
-	// worker runs. The workers stop, and are waited for, before the
+	// worker runs. The job workers and the notification worker stop, and
-	// database closes.
+	// are waited for, before the database closes.
-	jobsCtx, stopJobs := context.WithCancel(ctx)
+	workersCtx, stopWorkers := context.WithCancel(ctx)
 	defer func() {
-		stopJobs()
+		stopWorkers()
 		runner.Wait()
+		notifier.Wait()
 	}()
-	if err := runner.Start(jobsCtx); err != nil {
+	if err := runner.Start(workersCtx); err != nil {
+		return err
+	}
+	if err := notifier.Start(workersCtx); err != nil {
 		return err
 	}
 	// With those jobs failed, a dinner still pending with no grounding job

internal/db/notifications.sql.go

@@ -0,0 +1,202 @@
+// Code generated by sqlc. DO NOT EDIT.
+// versions:
+//   sqlc v1.31.1
+// source: notifications.sql
+
+package db
+
+import (
+	"context"
+	"database/sql"
+)
+
+const claimDueNotification = `-- name: ClaimDueNotification :one
+UPDATE notification_outbox
+SET next_attempt_at = ?1, updated_at = ?2
+WHERE id = (
+	SELECT id FROM notification_outbox
+	WHERE status = 'pending' AND next_attempt_at <= ?2
+	ORDER BY next_attempt_at, id
+	LIMIT 1
+)
+RETURNING id, event_key, channel, payload, status, attempts, next_attempt_at, last_error, created_at, updated_at
+`
+
+type ClaimDueNotificationParams struct {
+	LeaseUntil sql.NullString
+	Now        string
+}
+
+// Leases the pending notification due longest ago by moving its
+// next_attempt_at to lease_until, and returns it. A claimed row is not due
+// again until the lease ends, so no other claim takes it while it is being
+// delivered. Returns sql.ErrNoRows when none is due.
+func (q *Queries) ClaimDueNotification(ctx context.Context, arg ClaimDueNotificationParams) (NotificationOutbox, error) {
+	row := q.db.QueryRowContext(ctx, claimDueNotification, arg.LeaseUntil, arg.Now)
+	var i NotificationOutbox
+	err := row.Scan(
+		&i.ID,
+		&i.EventKey,
+		&i.Channel,
+		&i.Payload,
+		&i.Status,
+		&i.Attempts,
+		&i.NextAttemptAt,
+		&i.LastError,
+		&i.CreatedAt,
+		&i.UpdatedAt,
+	)
+	return i, err
+}
+
+const failNotification = `-- name: FailNotification :execrows
+UPDATE notification_outbox
+SET status = 'failed', attempts = attempts + ?1, next_attempt_at = NULL, last_error = ?2, updated_at = ?3
+WHERE id = ?4 AND status = 'pending' AND next_attempt_at = ?5
+`
+
+type FailNotificationParams struct {
+	Attempted  int64
+	LastError  sql.NullString
+	Now        string
+	ID         string
+	LeaseUntil sql.NullString
+}
+
+// Gives up on a notification. attempts counts the delivery attempt that
+// failed, if one was made. The lease must still be the claim's.
+func (q *Queries) FailNotification(ctx context.Context, arg FailNotificationParams) (int64, error) {
+	result, err := q.db.ExecContext(ctx, failNotification,
+		arg.Attempted,
+		arg.LastError,
+		arg.Now,
+		arg.ID,
+		arg.LeaseUntil,
+	)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
+const getNotificationPref = `-- name: GetNotificationPref :one
+SELECT event, push, chat, created_at, updated_at FROM notification_prefs
+WHERE event = ?
+`
+
+func (q *Queries) GetNotificationPref(ctx context.Context, event string) (NotificationPref, error) {
+	row := q.db.QueryRowContext(ctx, getNotificationPref, event)
+	var i NotificationPref
+	err := row.Scan(
+		&i.Event,
+		&i.Push,
+		&i.Chat,
+		&i.CreatedAt,
+		&i.UpdatedAt,
+	)
+	return i, err
+}
+
+const insertNotification = `-- name: InsertNotification :execrows
+INSERT INTO notification_outbox (
+	id, event_key, channel, payload, status, attempts, next_attempt_at, created_at, updated_at
+) VALUES (
+	?1, ?2, ?3, ?4, 'pending', 0, ?5, ?6, ?6
+)
+ON CONFLICT (event_key, channel) DO NOTHING
+`
+
+type InsertNotificationParams struct {
+	ID            string
+	EventKey      string
+	Channel       string
+	Payload       string
+	NextAttemptAt sql.NullString
+	Now           string
+}
+
+// Stores a pending notification for one channel, due at once. A row already
+// stored for the event key and channel, in any status, is left unchanged and
+// no row is inserted.
+func (q *Queries) InsertNotification(ctx context.Context, arg InsertNotificationParams) (int64, error) {
+	result, err := q.db.ExecContext(ctx, insertNotification,
+		arg.ID,
+		arg.EventKey,
+		arg.Channel,
+		arg.Payload,
+		arg.NextAttemptAt,
+		arg.Now,
+	)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
+const markNotificationSent = `-- name: MarkNotificationSent :execrows
+UPDATE notification_outbox
+SET status = 'sent', attempts = attempts + 1, next_attempt_at = NULL, updated_at = ?1
+WHERE id = ?2 AND status = 'pending' AND next_attempt_at = ?3
+`
+
+type MarkNotificationSentParams struct {
+	Now        string
+	ID         string
+	LeaseUntil sql.NullString
+}
+
+// Records a delivered notification; updated_at is when it was sent. The
+// lease must still be the claim's.
+func (q *Queries) MarkNotificationSent(ctx context.Context, arg MarkNotificationSentParams) (int64, error) {
+	result, err := q.db.ExecContext(ctx, markNotificationSent, arg.Now, arg.ID, arg.LeaseUntil)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
+const nextNotificationDue = `-- name: NextNotificationDue :one
+SELECT next_attempt_at FROM notification_outbox
+WHERE status = 'pending'
+ORDER BY next_attempt_at
+LIMIT 1
+`
+
+// Returns when the next pending notification is due. Returns sql.ErrNoRows
+// when none is pending.
+func (q *Queries) NextNotificationDue(ctx context.Context) (sql.NullString, error) {
+	row := q.db.QueryRowContext(ctx, nextNotificationDue)
+	var next_attempt_at sql.NullString
+	err := row.Scan(&next_attempt_at)
+	return next_attempt_at, err
+}
+
+const retryNotification = `-- name: RetryNotification :execrows
+UPDATE notification_outbox
+SET attempts = attempts + 1, next_attempt_at = ?1, last_error = ?2, updated_at = ?3
+WHERE id = ?4 AND status = 'pending' AND next_attempt_at = ?5
+`
+
+type RetryNotificationParams struct {
+	NextAttemptAt sql.NullString
+	LastError     sql.NullString
+	Now           string
+	ID            string
+	LeaseUntil    sql.NullString
+}
+
+// Records a failed delivery and when to try again. The lease must still be
+// the claim's.
+func (q *Queries) RetryNotification(ctx context.Context, arg RetryNotificationParams) (int64, error) {
+	result, err := q.db.ExecContext(ctx, retryNotification,
+		arg.NextAttemptAt,
+		arg.LastError,
+		arg.Now,
+		arg.ID,
+		arg.LeaseUntil,
+	)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}

internal/db/querier.go

@@ -12,6 +12,11 @@ import (
 type Querier interface {
 	// An unknown id returns sql.ErrNoRows.
 	ArchiveDinner(ctx context.Context, arg ArchiveDinnerParams) (Dinner, error)
+	// Leases the pending notification due longest ago by moving its
+	// next_attempt_at to lease_until, and returns it. A claimed row is not due
+	// again until the lease ends, so no other claim takes it while it is being
+	// delivered. Returns sql.ErrNoRows when none is due.
+	ClaimDueNotification(ctx context.Context, arg ClaimDueNotificationParams) (NotificationOutbox, error)
 	// Marks the oldest queued job running and returns it. Returns sql.ErrNoRows
 	// when no job is queued.
 	ClaimNextJob(ctx context.Context, now sql.NullString) (Job, error)
@@ -68,6 +73,9 @@ type Querier interface {
 	DeleteShoppingGroup(ctx context.Context, id string) error
 	DeleteShoppingItem(ctx context.Context, id string) error
 	FailJob(ctx context.Context, arg FailJobParams) (int64, error)
+	// Gives up on a notification. attempts counts the delivery attempt that
+	// failed, if one was made. The lease must still be the claim's.
+	FailNotification(ctx context.Context, arg FailNotificationParams) (int64, error)
 	// Fails every unarchived dinner still pending whose grounding job for its
 	// current grounding_revision is no longer queued or running: the process
 	// stopped before the job, or before its handler recorded a result. An
@@ -112,6 +120,7 @@ type Querier interface {
 	// The newest batch, whatever its status. Returns sql.ErrNoRows when there
 	// is none.
 	GetLatestSuggestionBatch(ctx context.Context) (SuggestionBatch, error)
+	GetNotificationPref(ctx context.Context, event string) (NotificationPref, error)
 	// The pending Rotation check of dinner_id. MarkTried makes one only for a
 	// dinner in the Quest log, which it moves to tried, so a dinner has at
 	// most one. Returns sql.ErrNoRows when there is none.
@@ -153,6 +162,10 @@ type Querier interface {
 	// them only when a member edited them while it was grounding. Like
 	// GroundDinner it clears the pasted recipe_url: the recipe replaces the link.
 	GroundPendingDinner(ctx context.Context, arg GroundPendingDinnerParams) (int64, error)
+	// Stores a pending notification for one channel, due at once. A row already
+	// stored for the event key and channel, in any status, is left unchanged and
+	// no row is inserted.
+	InsertNotification(ctx context.Context, arg InsertNotificationParams) (int64, error)
 	// The pending suggestions whose dinner is still suggested, oldest first,
 	// with their dinners.
 	ListDeck(ctx context.Context) ([]ListDeckRow, error)
@@ -211,6 +224,9 @@ type Querier interface {
 	// Moves a dinner still in the Quest log to tried; a dinner with any other
 	// status is left as it is (0 rows).
 	MarkDinnerTried(ctx context.Context, arg MarkDinnerTriedParams) (int64, error)
+	// Records a delivered notification; updated_at is when it was sent. The
+	// lease must still be the claim's.
+	MarkNotificationSent(ctx context.Context, arg MarkNotificationSentParams) (int64, error)
 	// Records an edit of plan id, raising its revision: a synced plan becomes
 	// changed, any other keeps its status.
 	MarkWeekPlanEdited(ctx context.Context, arg MarkWeekPlanEditedParams) error
@@ -230,6 +246,9 @@ type Querier interface {
 	// by a Rotation check decision, recording updated_by; its origin is kept.
 	// A dinner no longer tried is left as it is (0 rows).
 	MoveTriedDinner(ctx context.Context, arg MoveTriedDinnerParams) (int64, error)
+	// Returns when the next pending notification is due. Returns sql.ErrNoRows
+	// when none is pending.
+	NextNotificationDue(ctx context.Context) (sql.NullString, error)
 	// Stores a partial result of a running job.
 	PublishJobResult(ctx context.Context, arg PublishJobResultParams) (int64, error)
 	// Plans dinner_id on date of plan_id. An existing day keeps its calendar
@@ -241,6 +260,9 @@ type Querier interface {
 	RedraftWeekPlan(ctx context.Context, arg RedraftWeekPlanParams) error
 	// Records a generation of list id from plan plan_id.
 	RegenerateShoppingList(ctx context.Context, arg RegenerateShoppingListParams) error
+	// Records a failed delivery and when to try again. The lease must still be
+	// the claim's.
+	RetryNotification(ctx context.Context, arg RetryNotificationParams) (int64, error)
 	// An unknown id returns sql.ErrNoRows.
 	SetDinnerFavorite(ctx context.Context, arg SetDinnerFavoriteParams) (Dinner, error)
 	// Sets the chosen calendar; affects no row when nothing is connected.

internal/notify/notify.go

@@ -0,0 +1,248 @@
+// Package notify decides who is told what and delivers it reliably. It is
+// the only package that reads notification_prefs to choose channels.
+//
+// Emit is the one way to notify the household: it reads the preference row
+// of the event's type and, for each channel switched on, stores one pending
+// 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.
+//
+// The worker started by Start delivers due rows through the Sender
+// registered for their channel. A delivered row is marked sent. A failed
+// delivery is retried after a backoff of one minute doubling up to one hour,
+// until 24 hours after the row was created; the delivery that fails at or
+// after that point marks it failed, with the error in last_error.
+package notify
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"net/url"
+	"strings"
+	"sync"
+	"time"
+
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/db"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store"
+)
+
+// EventType is what a notification is about. Each has one row in
+// notification_prefs saying which channels it goes to.
+type EventType string
+
+// The event types, as notification_prefs.event names them.
+const (
+	PlanReady      EventType = "plan_ready"
+	ListReady      EventType = "list_ready"
+	DayOfReminder  EventType = "day_of_reminder"
+	NewSuggestions EventType = "new_suggestions"
+	RotationReview EventType = "rotation_review"
+)
+
+// eventTypes is every EventType.
+var eventTypes = map[EventType]bool{
+	PlanReady: true, ListReady: true, DayOfReminder: true, NewSuggestions: true, RotationReview: true,
+}
+
+// Channel is a way of telling the household, as notification_outbox.channel
+// names it.
+type Channel string
+
+// The channels.
+const (
+	ChannelPush Channel = "push"
+	ChannelChat Channel = "chat"
+)
+
+// Event is something to tell the household.
+type Event struct {
+	// Type is what the event is about.
+	Type EventType
+	// Key identifies this event: it starts with Type and a colon, as in
+	// "plan_ready:2026-10-12". An event emitted again with the same key is
+	// ignored.
+	Key string
+	// Title is the notification's headline; it must not be blank.
+	Title string
+	// Body is the notification's text; it may be empty.
+	Body string
+	// URL is the path-absolute, same-origin path a tap on the notification
+	// opens, such as "/plan".
+	URL string
+	// Week is the Monday of the week the event concerns, for the senders'
+	// message builders; zero when the event concerns no week.
+	Week store.LocalDate
+}
+
+// Payload is the JSON stored in notification_outbox.payload and handed to
+// a Sender: the one shape every channel delivers from. Its title, body, tag
+// and url are the notice the service worker shows (web/src/push/notification.ts),
+// which ignores the other fields.
+type Payload struct {
+	Event EventType `json:"event"`
+	Title string    `json:"title"`
+	Body  string    `json:"body"`
+	// URL is a path-absolute, same-origin path. A sender that needs an
+	// absolute URL resolves it against the public URL.
+	URL string `json:"url"`
+	// Tag is the event key, so a notification shown again for the same
+	// event replaces the first.
+	Tag string `json:"tag"`
+	// Week is the YYYY-MM-DD Monday of the event's week, or absent.
+	Week string `json:"week,omitempty"`
+}
+
+// ErrInvalidEvent is returned by Emit for an event it refuses; nothing is
+// stored.
+var ErrInvalidEvent = errors.New("notify: invalid event")
+
+// Service emits notifications and runs the worker that delivers them.
+type Service struct {
+	st     *store.Store
+	logger *slog.Logger
+	// now is the clock rows are created, claimed and retried on; tests
+	// replace it.
+	now func() time.Time
+	// wake holds up to one token; Emit adds one so the worker looks for
+	// the rows it stored.
+	wake chan struct{}
+
+	mu      sync.Mutex
+	senders map[Channel]Sender
+	started bool
+
+	wg sync.WaitGroup
+}
+
+// NewService returns a service over st with no senders registered.
+func NewService(st *store.Store, logger *slog.Logger) (*Service, error) {
+	if st == nil {
+		return nil, errors.New("notify: nil store")
+	}
+	if logger == nil {
+		return nil, errors.New("notify: nil logger")
+	}
+	return &Service{
+		st:      st,
+		logger:  logger,
+		now:     time.Now,
+		wake:    make(chan struct{}, 1),
+		senders: make(map[Channel]Sender),
+	}, nil
+}
+
+// Emit stores one pending notification of ev for each channel its type's
+// preference switches on, and wakes the worker. A channel already holding a
+// 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 {
+	if err := validate(ev); err != nil {
+		return err
+	}
+	payload := Payload{Event: ev.Type, Title: ev.Title, Body: ev.Body, URL: ev.URL, Tag: ev.Key}
+	if !ev.Week.IsZero() {
+		payload.Week = ev.Week.String()
+	}
+	data, err := json.Marshal(payload)
+	if err != nil {
+		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
+	})
+	if err != nil {
+		return err
+	}
+	s.logger.InfoContext(ctx, "notification emitted", "event_key", ev.Key, "queued", queued)
+	if len(queued) > 0 {
+		s.wakeWorker()
+	}
+	return nil
+}
+
+// enabledChannels returns the channels pref switches on.
+func enabledChannels(pref db.NotificationPref) []Channel {
+	var chs []Channel
+	if pref.Push == 1 {
+		chs = append(chs, ChannelPush)
+	}
+	if pref.Chat == 1 {
+		chs = append(chs, ChannelChat)
+	}
+	return chs
+}
+
+// validate reports why ev cannot be emitted, wrapping ErrInvalidEvent.
+func validate(ev Event) error {
+	if !eventTypes[ev.Type] {
+		return fmt.Errorf("%w: unknown type %q", ErrInvalidEvent, ev.Type)
+	}
+	prefix := string(ev.Type) + ":"
+	if !strings.HasPrefix(ev.Key, prefix) || len(ev.Key) == len(prefix) {
+		return fmt.Errorf("%w: key %q does not start with %q and an identifier", ErrInvalidEvent, ev.Key, prefix)
+	}
+	if strings.TrimSpace(ev.Title) == "" {
+		return fmt.Errorf("%w: %s has a blank title", ErrInvalidEvent, ev.Key)
+	}
+	if !isAppPath(ev.URL) {
+		return fmt.Errorf("%w: %s url %q is not a path-absolute same-origin path", ErrInvalidEvent, ev.Key, ev.URL)
+	}
+	return nil
+}
+
+// isAppPath reports whether raw is a path on the app's own origin, as the
+// service worker requires: it starts with one slash and names no scheme or
+// host, however it is parsed.
+func isAppPath(raw string) bool {
+	if !strings.HasPrefix(raw, "/") || strings.HasPrefix(raw, "//") || strings.ContainsAny(raw, "\\\t\r\n") {
+		return false
+	}
+	u, err := url.Parse(raw)
+	return err == nil && u.Scheme == "" && u.Host == "" && !strings.HasPrefix(u.Path, "//")
+}
+
+// timestamp returns t in the stored UTC RFC 3339 form.
+func timestamp(t time.Time) string {
+	return store.Timestamp{Time: t.UTC().Truncate(time.Millisecond)}.String()
+}
+
+// parseTimestamp parses a stored timestamp.
+func parseTimestamp(s string) (time.Time, error) {
+	var ts store.Timestamp
+	if err := ts.Scan(s); err != nil {
+		return time.Time{}, err
+	}
+	return ts.Time, nil
+}

internal/notify/notify_test.go

@@ -0,0 +1,730 @@
+package notify
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"sync"
+	"testing"
+	"time"
+
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/db"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/household"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/migrations"
+)
+
+// start is the fake clock's first instant.
+var start = time.Date(2026, 10, 9, 15, 0, 0, 0, time.UTC)
+
+// clock is a controllable clock.
+type clock struct {
+	mu sync.Mutex
+	t  time.Time
+}
+
+func (c *clock) now() time.Time {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.t
+}
+
+func (c *clock) set(t time.Time) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.t = t
+}
+
+// fakeSender records the messages it is given and fails while fail is set.
+type fakeSender struct {
+	mu   sync.Mutex
+	fail error
+	got  []Message
+}
+
+func (f *fakeSender) Send(_ context.Context, m Message) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.got = append(f.got, m)
+	return f.fail
+}
+
+func (f *fakeSender) messages() []Message {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([]Message(nil), f.got...)
+}
+
+func newTestService(t *testing.T) (*Service, *store.Store, *clock) {
+	t.Helper()
+	ctx := context.Background()
+	st, err := store.Open(ctx, t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if err := st.Close(); err != nil {
+			t.Error(err)
+		}
+	})
+	if err := store.Migrate(ctx, st.Writer, migrations.FS); err != nil {
+		t.Fatal(err)
+	}
+	if err := household.SeedNotificationPrefs(ctx, st); err != nil {
+		t.Fatal(err)
+	}
+	svc, err := NewService(st, slog.New(slog.NewTextHandler(io.Discard, nil)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	c := &clock{t: start}
+	svc.now = c.now
+	return svc, st, c
+}
+
+func setPref(t *testing.T, st *store.Store, ev EventType, push, chat bool) {
+	t.Helper()
+	res, err := st.Writer.Exec(`UPDATE notification_prefs SET push = ?, chat = ? WHERE event = ?`, push, chat, string(ev))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n, _ := res.RowsAffected(); n != 1 {
+		t.Fatalf("no notification preference for %s", ev)
+	}
+}
+
+// outbox returns the stored rows of key by channel.
+func outbox(t *testing.T, st *store.Store, key string) map[Channel]db.NotificationOutbox {
+	t.Helper()
+	rs, err := st.Reader.Query(`SELECT id, event_key, channel, payload, status, attempts, next_attempt_at, last_error, created_at, updated_at
+		FROM notification_outbox WHERE event_key = ?`, key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rs.Close()
+	got := map[Channel]db.NotificationOutbox{}
+	for rs.Next() {
+		var r db.NotificationOutbox
+		if err := rs.Scan(&r.ID, &r.EventKey, &r.Channel, &r.Payload, &r.Status, &r.Attempts, &r.NextAttemptAt, &r.LastError, &r.CreatedAt, &r.UpdatedAt); err != nil {
+			t.Fatal(err)
+		}
+		if _, dup := got[Channel(r.Channel)]; dup {
+			t.Fatalf("two %s rows for %s", r.Channel, key)
+		}
+		got[Channel(r.Channel)] = r
+	}
+	if err := rs.Err(); err != nil {
+		t.Fatal(err)
+	}
+	return got
+}
+
+func onlyRow(t *testing.T, st *store.Store, key string, ch Channel) db.NotificationOutbox {
+	t.Helper()
+	rows := outbox(t, st, key)
+	r, ok := rows[ch]
+	if !ok || len(rows) != 1 {
+		t.Fatalf("rows of %s = %v, want one %s row", key, rows, ch)
+	}
+	return r
+}
+
+func planReady(key string) Event {
+	return Event{
+		Type:  PlanReady,
+		Key:   key,
+		Title: "This week's plan is ready",
+		Body:  "Five dinners on the Draft board.",
+		URL:   "/plan",
+		Week:  mustDate("2026-10-12"),
+	}
+}
+
+func mustDate(s string) store.LocalDate {
+	d, err := store.ParseLocalDate(s)
+	if err != nil {
+		panic(err)
+	}
+	return d
+}
+
+func TestEmitRespectsPrefsMatrix(t *testing.T) {
+	ctx := context.Background()
+	for _, tc := range []struct {
+		push, chat bool
+		want       []Channel
+	}{
+		{push: false, chat: false, want: nil},
+		{push: true, chat: false, want: []Channel{ChannelPush}},
+		{push: false, chat: true, want: []Channel{ChannelChat}},
+		{push: true, chat: true, want: []Channel{ChannelPush, ChannelChat}},
+	} {
+		t.Run(fmt.Sprintf("push=%v,chat=%v", tc.push, tc.chat), func(t *testing.T) {
+			svc, st, _ := newTestService(t)
+			for ev := range eventTypes {
+				setPref(t, st, ev, tc.push, tc.chat)
+				key := string(ev) + ":2026-10-12"
+				if err := svc.Emit(ctx, Event{Type: ev, Key: key, Title: "Title", Body: "Body", URL: "/"}); err != nil {
+					t.Fatal(err)
+				}
+				rows := outbox(t, st, key)
+				if len(rows) != len(tc.want) {
+					t.Fatalf("%s: %d rows, want channels %v", ev, len(rows), tc.want)
+				}
+				for _, ch := range tc.want {
+					r, ok := rows[ch]
+					if !ok {
+						t.Fatalf("%s: no %s row", ev, ch)
+					}
+					now := timestamp(start)
+					if r.Status != "pending" || r.Attempts != 0 || r.NextAttemptAt.String != now || r.CreatedAt != now || r.LastError.Valid {
+						t.Errorf("%s %s row = %+v, want pending, due now", ev, ch, r)
+					}
+				}
+			}
+		})
+	}
+}
+
+func TestEmitStoresPayload(t *testing.T) {
+	svc, st, _ := newTestService(t)
+	setPref(t, st, PlanReady, true, true)
+	if err := svc.Emit(context.Background(), planReady("plan_ready:2026-10-12")); err != nil {
+		t.Fatal(err)
+	}
+	want := `{"event":"plan_ready","title":"This week's plan is ready","body":"Five dinners on the Draft board.","url":"/plan","tag":"plan_ready:2026-10-12","week":"2026-10-12"}`
+	for ch, r := range outbox(t, st, "plan_ready:2026-10-12") {
+		if r.Payload != want {
+			t.Errorf("%s payload = %s, want %s", ch, r.Payload, want)
+		}
+	}
+
+	// An event about no week omits it.
+	if err := svc.Emit(context.Background(), Event{Type: NewSuggestions, Key: "new_suggestions:batch1", Title: "New Side quests", URL: "/suggestions"}); err != nil {
+		t.Fatal(err)
+	}
+	r := onlyRow(t, st, "new_suggestions:batch1", ChannelPush)
+	if want := `{"event":"new_suggestions","title":"New Side quests","body":"","url":"/suggestions","tag":"new_suggestions:batch1"}`; r.Payload != want {
+		t.Errorf("payload = %s, want %s", r.Payload, want)
+	}
+}
+
+func TestEmitIgnoresDuplicateKey(t *testing.T) {
+	ctx := context.Background()
+	svc, st, c := newTestService(t)
+	setPref(t, st, PlanReady, true, false)
+	const key = "plan_ready:2026-10-12"
+	if err := svc.Emit(ctx, planReady(key)); err != nil {
+		t.Fatal(err)
+	}
+	first := onlyRow(t, st, key, ChannelPush)
+
+	c.set(start.Add(time.Minute))
+	again := planReady(key)
+	again.Title = "A different title"
+	if err := svc.Emit(ctx, again); err != nil {
+		t.Fatalf("duplicate emit: %v", err)
+	}
+	if got := onlyRow(t, st, key, ChannelPush); got != first {
+		t.Fatalf("duplicate emit changed the row:\n got %+v\nwant %+v", got, first)
+	}
+
+	// Once sent, a duplicate still leaves the row as it is.
+	sender := &fakeSender{}
+	if err := svc.Register(ChannelPush, sender); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := svc.deliverDue(ctx); err != nil {
+		t.Fatal(err)
+	}
+	sent := onlyRow(t, st, key, ChannelPush)
+	if sent.Status != "sent" {
+		t.Fatalf("status = %s, want sent", sent.Status)
+	}
+	c.set(start.Add(2 * time.Minute))
+	if err := svc.Emit(ctx, again); err != nil {
+		t.Fatalf("emit after sent: %v", err)
+	}
+	if got := onlyRow(t, st, key, ChannelPush); got != sent {
+		t.Fatalf("emit after sent changed the row:\n got %+v\nwant %+v", got, sent)
+	}
+	if _, err := svc.deliverDue(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if n := len(sender.messages()); n != 1 {
+		t.Fatalf("%d deliveries, want 1", n)
+	}
+}
+
+func TestEmitRejectsInvalidEvents(t *testing.T) {
+	svc, st, _ := newTestService(t)
+	valid := planReady("plan_ready:2026-10-12")
+	for name, mutate := range map[string]func(*Event){
+		"unknown type":         func(e *Event) { e.Type = "dinner_burnt" },
+		"empty key":            func(e *Event) { e.Key = "" },
+		"key without id":       func(e *Event) { e.Key = "plan_ready:" },
+		"key of another type":  func(e *Event) { e.Key = "list_ready:2026-10-12" },
+		"blank title":          func(e *Event) { e.Title = "  " },
+		"empty url":            func(e *Event) { e.URL = "" },
+		"relative url":         func(e *Event) { e.URL = "plan" },
+		"absolute url":         func(e *Event) { e.URL = "https://evil.example/plan" },
+		"protocol-relative":    func(e *Event) { e.URL = "//evil.example/plan" },
+		"backslash url":        func(e *Event) { e.URL = "/\\evil.example" },
+		"url with a tab":       func(e *Event) { e.URL = "/\t/evil.example" },
+		"url with a bad query": func(e *Event) { e.URL = "/plan%zz" },
+	} {
+		t.Run(name, func(t *testing.T) {
+			ev := valid
+			mutate(&ev)
+			if err := svc.Emit(context.Background(), ev); !errors.Is(err, ErrInvalidEvent) {
+				t.Fatalf("Emit = %v, want ErrInvalidEvent", err)
+			}
+		})
+	}
+	var n int
+	if err := st.Reader.QueryRow(`SELECT COUNT(*) FROM notification_outbox`).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	if n != 0 {
+		t.Fatalf("%d rows stored, want 0", n)
+	}
+}
+
+func TestEmitWithoutPrefIsAnError(t *testing.T) {
+	svc, st, _ := newTestService(t)
+	if _, err := st.Writer.Exec(`DELETE FROM notification_prefs WHERE event = 'plan_ready'`); err != nil {
+		t.Fatal(err)
+	}
+	err := svc.Emit(context.Background(), planReady("plan_ready:2026-10-12"))
+	if err == nil || errors.Is(err, ErrInvalidEvent) {
+		t.Fatalf("Emit = %v, want a missing-preference error", err)
+	}
+}
+
+func TestDeliveryMarkedSent(t *testing.T) {
+	ctx := context.Background()
+	svc, st, c := newTestService(t)
+	setPref(t, st, PlanReady, true, true)
+	push, chat := &fakeSender{}, &fakeSender{}
+	if err := svc.Register(ChannelPush, push); err != nil {
+		t.Fatal(err)
+	}
+	if err := svc.Register(ChannelChat, chat); err != nil {
+		t.Fatal(err)
+	}
+	const key = "plan_ready:2026-10-12"
+	if err := svc.Emit(ctx, planReady(key)); err != nil {
+		t.Fatal(err)
+	}
+	sentAt := start.Add(3 * time.Second)
+	c.set(sentAt)
+	wait, err := svc.deliverDue(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if wait != -1 {
+		t.Errorf("wait = %v, want -1 with nothing pending", wait)
+	}
+
+	rows := outbox(t, st, key)
+	wantPayload := Payload{Event: PlanReady, Title: "This week's plan is ready", Body: "Five dinners on the Draft board.", URL: "/plan", Tag: key, Week: "2026-10-12"}
+	for ch, sender := range map[Channel]*fakeSender{ChannelPush: push, ChannelChat: chat} {
+		r := rows[ch]
+		if r.Status != "sent" || r.Attempts != 1 || r.NextAttemptAt.Valid || r.UpdatedAt != timestamp(sentAt) || r.LastError.Valid {
+			t.Errorf("%s row = %+v, want sent at %s after 1 attempt", ch, r, timestamp(sentAt))
+		}
+		got := sender.messages()
+		want := Message{ID: r.ID, EventKey: key, Channel: ch, Payload: wantPayload, Attempt: 1}
+		if len(got) != 1 || got[0] != want {
+			t.Errorf("%s sender got %+v, want [%+v]", ch, got, want)
+		}
+	}
+}
+
+func TestRetryBackoffSchedule(t *testing.T) {
+	ctx := context.Background()
+	svc, st, c := newTestService(t)
+	setPref(t, st, PlanReady, true, false)
+	sender := &fakeSender{fail: errors.New("push service unavailable")}
+	if err := svc.Register(ChannelPush, sender); err != nil {
+		t.Fatal(err)
+	}
+	const key = "plan_ready:2026-10-12"
+	if err := svc.Emit(ctx, planReady(key)); err != nil {
+		t.Fatal(err)
+	}
+
+	now := start
+	for i, wantDelay := range []time.Duration{
+		time.Minute, 2 * time.Minute, 4 * time.Minute, 8 * time.Minute, 16 * time.Minute,
+		32 * time.Minute, time.Hour, time.Hour, time.Hour,
+	} {
+		c.set(now)
+		wait, err := svc.deliverDue(ctx)
+		if err != nil {
+			t.Fatal(err)
+		}
+		r := onlyRow(t, st, key, ChannelPush)
+		wantNext := now.Add(wantDelay)
+		if r.Status != "pending" || r.Attempts != int64(i+1) || r.NextAttemptAt.String != timestamp(wantNext) ||
+			r.LastError.String != "push service unavailable" {
+			t.Fatalf("after failure %d row = %+v, want pending, attempts %d, next %s", i+1, r, i+1, timestamp(wantNext))
+		}
+		if wait != wantDelay {
+			t.Fatalf("after failure %d wait = %v, want %v", i+1, wait, wantDelay)
+		}
+
+		// Not due a moment before the retry time.
+		c.set(wantNext.Add(-time.Millisecond))
+		if _, err := svc.deliverDue(ctx); err != nil {
+			t.Fatal(err)
+		}
+		if n := len(sender.messages()); n != i+1 {
+			t.Fatalf("%d deliveries before the retry was due, want %d", n, i+1)
+		}
+		now = wantNext
+	}
+	for i, m := range sender.messages() {
+		if m.Attempt != int64(i+1) {
+			t.Errorf("delivery %d has Attempt %d", i+1, m.Attempt)
+		}
+	}
+}
+
+func TestRetryThenSent(t *testing.T) {
+	ctx := context.Background()
+	svc, st, c := newTestService(t)
+	setPref(t, st, ListReady, false, true)
+	sender := &fakeSender{fail: errors.New("chat returned 503")}
+	if err := svc.Register(ChannelChat, sender); err != nil {
+		t.Fatal(err)
+	}
+	const key = "list_ready:2026-10-12"
+	if err := svc.Emit(ctx, Event{Type: ListReady, Key: key, Title: "Supply run ready", URL: "/supply-run"}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := svc.deliverDue(ctx); err != nil {
+		t.Fatal(err)
+	}
+	sender.mu.Lock()
+	sender.fail = nil
+	sender.mu.Unlock()
+	c.set(start.Add(time.Minute))
+	if _, err := svc.deliverDue(ctx); err != nil {
+		t.Fatal(err)
+	}
+	r := onlyRow(t, st, key, ChannelChat)
+	if r.Status != "sent" || r.Attempts != 2 || r.UpdatedAt != timestamp(start.Add(time.Minute)) {
+		t.Fatalf("row = %+v, want sent on attempt 2", r)
+	}
+}
+
+func TestGiveUpAfter24Hours(t *testing.T) {
+	ctx := context.Background()
+	svc, st, c := newTestService(t)
+	setPref(t, st, DayOfReminder, true, false)
+	sender := &fakeSender{fail: errors.New("push service unavailable")}
+	if err := svc.Register(ChannelPush, sender); err != nil {
+		t.Fatal(err)
+	}
+	const key = "day_of_reminder:2026-10-09"
+	if err := svc.Emit(ctx, Event{Type: DayOfReminder, Key: key, Title: "Tonight: tacos", URL: "/"}); err != nil {
+		t.Fatal(err)
+	}
+
+	// Deliver each retry as it falls due until the row leaves pending.
+	deadline := start.Add(24 * time.Hour)
+	var times []time.Time
+	for range 100 {
+		r := onlyRow(t, st, key, ChannelPush)
+		if r.Status != "pending" {
+			break
+		}
+		due, err := parseTimestamp(r.NextAttemptAt.String)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if due.After(deadline) {
+			t.Fatalf("retry due %s, after the 24 h deadline %s", due, deadline)
+		}
+		c.set(due)
+		times = append(times, due)
+		if _, err := svc.deliverDue(ctx); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	// Attempts at 0, 1, 3, 7, 15, 31 and 63 minutes, then hourly to 1383
+	// minutes (22 more): 29 in all. The row is due once more at the
+	// deadline, where it is failed without being sent.
+	const wantAttempts = 29
+	r := onlyRow(t, st, key, ChannelPush)
+	if r.Status != "failed" || r.Attempts != wantAttempts || r.NextAttemptAt.Valid ||
+		r.LastError.String != "not delivered within 24 hours of creation; last error: push service unavailable" ||
+		r.UpdatedAt != timestamp(deadline) {
+		t.Fatalf("row = %+v, want failed at %s after %d attempts", r, timestamp(deadline), wantAttempts)
+	}
+	if len(times) != wantAttempts+1 || !times[len(times)-1].Equal(deadline) || !times[len(times)-2].Equal(start.Add(1383*time.Minute)) {
+		t.Fatalf("due at %v, want %d attempts ending 1383 min after creation, then the deadline", times, wantAttempts)
+	}
+	if n := len(sender.messages()); n != wantAttempts {
+		t.Fatalf("%d deliveries, want %d", n, wantAttempts)
+	}
+
+	// A failed row is never delivered again.
+	c.set(deadline.Add(48 * time.Hour))
+	if _, err := svc.deliverDue(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if n := len(sender.messages()); n != wantAttempts {
+		t.Fatalf("%d deliveries, want %d", n, wantAttempts)
+	}
+}
+
+// TestOverdueRowIsNeverDelivered covers a row first claimed 24 hours or
+// more after its creation, as when the service was down: it is failed
+// without calling the sender, whether the sender would succeed or fail.
+func TestOverdueRowIsNeverDelivered(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		at   time.Duration
+		fail error
+	}{
+		{"succeeding sender at 24 h", 24 * time.Hour, nil},
+		{"succeeding sender after 24 h", 25 * time.Hour, nil},
+		{"failing sender at 24 h", 24 * time.Hour, errors.New("gone")},
+		{"failing sender after 24 h", 25 * time.Hour, errors.New("gone")},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			svc, st, c := newTestService(t)
+			setPref(t, st, PlanReady, true, false)
+			sender := &fakeSender{fail: tc.fail}
+			if err := svc.Register(ChannelPush, sender); err != nil {
+				t.Fatal(err)
+			}
+			const key = "plan_ready:2026-10-12"
+			if err := svc.Emit(ctx, planReady(key)); err != nil {
+				t.Fatal(err)
+			}
+			c.set(start.Add(tc.at))
+			if _, err := svc.deliverDue(ctx); err != nil {
+				t.Fatal(err)
+			}
+			if n := len(sender.messages()); n != 0 {
+				t.Fatalf("%d deliveries of an overdue row, want 0", n)
+			}
+			r := onlyRow(t, st, key, ChannelPush)
+			if r.Status != "failed" || r.Attempts != 0 || r.NextAttemptAt.Valid ||
+				r.LastError.String != "not delivered within 24 hours of creation" || r.UpdatedAt != timestamp(start.Add(tc.at)) {
+				t.Fatalf("row = %+v, want failed without an attempt", r)
+			}
+		})
+	}
+}
+
+// TestRowJustBeforeDeadlineIsDelivered pins the boundary from the other
+// side: a row claimed just under 24 hours after its creation is sent.
+func TestRowJustBeforeDeadlineIsDelivered(t *testing.T) {
+	ctx := context.Background()
+	svc, st, c := newTestService(t)
+	setPref(t, st, PlanReady, true, false)
+	sender := &fakeSender{}
+	if err := svc.Register(ChannelPush, sender); err != nil {
+		t.Fatal(err)
+	}
+	const key = "plan_ready:2026-10-12"
+	if err := svc.Emit(ctx, planReady(key)); err != nil {
+		t.Fatal(err)
+	}
+	c.set(start.Add(24*time.Hour - time.Millisecond))
+	if _, err := svc.deliverDue(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if r := onlyRow(t, st, key, ChannelPush); r.Status != "sent" || r.Attempts != 1 || len(sender.messages()) != 1 {
+		t.Fatalf("row = %+v after %d deliveries, want sent once", r, len(sender.messages()))
+	}
+}
+
+func TestChannelWithoutSenderFails(t *testing.T) {
+	ctx := context.Background()
+	svc, st, _ := newTestService(t)
+	setPref(t, st, PlanReady, true, true)
+	push := &fakeSender{}
+	if err := svc.Register(ChannelPush, push); err != nil {
+		t.Fatal(err)
+	}
+	const key = "plan_ready:2026-10-12"
+	if err := svc.Emit(ctx, planReady(key)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := svc.deliverDue(ctx); err != nil {
+		t.Fatal(err)
+	}
+	rows := outbox(t, st, key)
+	if r := rows[ChannelChat]; r.Status != "failed" || r.Attempts != 0 || r.LastError.String != "no sender registered for channel chat" {
+		t.Fatalf("chat row = %+v, want failed with no sender", r)
+	}
+	if r := rows[ChannelPush]; r.Status != "sent" {
+		t.Fatalf("push row = %+v, want sent", r)
+	}
+}
+
+func TestLeasedRowIsNotClaimedAgain(t *testing.T) {
+	ctx := context.Background()
+	svc, st, c := newTestService(t)
+	setPref(t, st, PlanReady, true, false)
+	sender := &fakeSender{}
+	if err := svc.Register(ChannelPush, sender); err != nil {
+		t.Fatal(err)
+	}
+	const key = "plan_ready:2026-10-12"
+	if err := svc.Emit(ctx, planReady(key)); err != nil {
+		t.Fatal(err)
+	}
+	// Another tick claims the row and stops before recording the delivery.
+	lease := start.Add(leaseFor)
+	if _, err := db.New(st.Writer).ClaimDueNotification(ctx, db.ClaimDueNotificationParams{
+		LeaseUntil: sql.NullString{String: timestamp(lease), Valid: true}, Now: timestamp(start),
+	}); err != nil {
+		t.Fatal(err)
+	}
+	c.set(lease.Add(-time.Millisecond))
+	if _, err := svc.deliverDue(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if n := len(sender.messages()); n != 0 {
+		t.Fatalf("%d deliveries of a leased row, want 0", n)
+	}
+	// Once the lease ends the row is delivered.
+	c.set(lease)
+	if _, err := svc.deliverDue(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if r := onlyRow(t, st, key, ChannelPush); r.Status != "sent" || r.Attempts != 1 || len(sender.messages()) != 1 {
+		t.Fatalf("row = %+v after %d deliveries, want sent once", r, len(sender.messages()))
+	}
+}
+
+// blockingSender holds every Send until release is closed.
+type blockingSender struct {
+	fakeSender
+	entered chan struct{}
+	release chan struct{}
+}
+
+func (b *blockingSender) Send(ctx context.Context, m Message) error {
+	b.entered <- struct{}{}
+	<-b.release
+	return b.fakeSender.Send(ctx, m)
+}
+
+func TestConcurrentTicksDeliverOnce(t *testing.T) {
+	ctx := context.Background()
+	svc, st, _ := newTestService(t)
+	setPref(t, st, PlanReady, true, false)
+	sender := &blockingSender{entered: make(chan struct{}, 10), release: make(chan struct{})}
+	if err := svc.Register(ChannelPush, sender); err != nil {
+		t.Fatal(err)
+	}
+	const key = "plan_ready:2026-10-12"
+	if err := svc.Emit(ctx, planReady(key)); err != nil {
+		t.Fatal(err)
+	}
+
+	var wg sync.WaitGroup
+	first := make(chan error, 1)
+	wg.Go(func() {
+		_, err := svc.deliverDue(ctx)
+		first <- err
+	})
+	<-sender.entered // the first tick holds the row
+	if _, err := svc.deliverDue(ctx); err != nil {
+		t.Fatal(err)
+	}
+	close(sender.release)
+	wg.Wait()
+	if err := <-first; err != nil {
+		t.Fatal(err)
+	}
+	if n := len(sender.messages()); n != 1 {
+		t.Fatalf("%d deliveries, want 1", n)
+	}
+	if r := onlyRow(t, st, key, ChannelPush); r.Status != "sent" {
+		t.Fatalf("row = %+v, want sent", r)
+	}
+}
+
+func TestRetryDelay(t *testing.T) {
+	for attempt, want := range map[int64]time.Duration{
+		1: time.Minute, 2: 2 * time.Minute, 3: 4 * time.Minute, 6: 32 * time.Minute,
+		7: time.Hour, 8: time.Hour, 1000: time.Hour,
+	} {
+		if got := retryDelay(attempt); got != want {
+			t.Errorf("retryDelay(%d) = %v, want %v", attempt, got, want)
+		}
+	}
+}
+
+func TestRegister(t *testing.T) {
+	svc, _, _ := newTestService(t)
+	if err := svc.Register("email", &fakeSender{}); err == nil {
+		t.Error("registered an unknown channel")
+	}
+	if err := svc.Register(ChannelPush, nil); err == nil {
+		t.Error("registered a nil sender")
+	}
+	if err := svc.Register(ChannelPush, &fakeSender{}); err != nil {
+		t.Fatal(err)
+	}
+	if err := svc.Register(ChannelPush, &fakeSender{}); err == nil {
+		t.Error("registered a second push sender")
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	if err := svc.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if err := svc.Register(ChannelChat, &fakeSender{}); err == nil {
+		t.Error("registered a sender after Start")
+	}
+	if err := svc.Start(ctx); err == nil {
+		t.Error("started twice")
+	}
+	cancel()
+	svc.Wait()
+}
+
+// TestWorkerDeliversEmittedEvent runs the worker on the real clock: Emit
+// wakes it, it delivers, and it stops when its context is cancelled.
+func TestWorkerDeliversEmittedEvent(t *testing.T) {
+	svc, st, _ := newTestService(t)
+	svc.now = time.Now
+	setPref(t, st, PlanReady, true, false)
+	sender := &fakeSender{}
+	if err := svc.Register(ChannelPush, sender); err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	if err := svc.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	const key = "plan_ready:2026-10-12"
+	if err := svc.Emit(context.Background(), planReady(key)); err != nil {
+		t.Fatal(err)
+	}
+	deadline := time.Now().Add(5 * time.Second)
+	for onlyRow(t, st, key, ChannelPush).Status != "sent" {
+		if time.Now().After(deadline) {
+			t.Fatal("worker did not deliver the notification")
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	cancel()
+	svc.Wait()
+	if n := len(sender.messages()); n != 1 {
+		t.Fatalf("%d deliveries, want 1", n)
+	}
+}

internal/notify/worker.go

@@ -0,0 +1,299 @@
+package notify
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/db"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
+)
+
+// Delivery timing.
+const (
+	// retryFirst is the wait after a row's first failed delivery; each
+	// further failure doubles it, up to retryMax.
+	retryFirst = time.Minute
+	retryMax   = time.Hour
+	// giveUpAfter is how long after its creation a row may be delivered.
+	// The last retry is due at that moment; a row claimed then or later is
+	// marked failed without being sent, and a delivery failing at or
+	// after it marks the row failed.
+	giveUpAfter = 24 * time.Hour
+	// sendTimeout bounds one Send.
+	sendTimeout = 30 * time.Second
+	// leaseFor is how long a claimed row is not due again. It outlasts
+	// sendTimeout and the final write, so a row is only claimed again when
+	// the process that claimed it stopped before recording the delivery.
+	leaseFor = 2 * time.Minute
+	// pollRetryMin and pollRetryMax bound the backoff after a database
+	// error while claiming rows.
+	pollRetryMin = time.Second
+	pollRetryMax = time.Minute
+)
+
+// Message is one notification for a Sender to deliver.
+type Message struct {
+	// ID is the notification_outbox row id.
+	ID string
+	// EventKey is the event's idempotency key.
+	EventKey string
+	Channel  Channel
+	Payload  Payload
+	// Attempt is this delivery's number, from 1.
+	Attempt int64
+}
+
+// Sender delivers notifications on one channel. Send returns nil once the
+// channel has accepted m, and an error otherwise; the worker retries it.
+// The error's text is stored in notification_outbox.last_error and logged,
+// so it must never carry a secret: no webhook URL, key or token (a
+// *url.Error quotes its URL).
+type Sender interface {
+	Send(ctx context.Context, m Message) error
+}
+
+// Register makes sender the Sender of ch. It must be called before Start,
+// once per channel.
+func (s *Service) Register(ch Channel, sender Sender) error {
+	if ch != ChannelPush && ch != ChannelChat {
+		return fmt.Errorf("notify: register: unknown channel %q", ch)
+	}
+	if sender == nil {
+		return fmt.Errorf("notify: register %s: nil sender", ch)
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.started {
+		return fmt.Errorf("notify: register %s: worker already started", ch)
+	}
+	if _, ok := s.senders[ch]; ok {
+		return fmt.Errorf("notify: register %s: channel already has a sender", ch)
+	}
+	s.senders[ch] = sender
+	return nil
+}
+
+// Start starts the worker. It stops when ctx is cancelled; Wait waits for
+// it. A row whose delivery is cut short by ctx keeps its lease and is
+// delivered again once the lease ends.
+func (s *Service) Start(ctx context.Context) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.started {
+		return errors.New("notify: worker already started")
+	}
+	s.started = true
+	s.wg.Add(1)
+	go s.work(ctx)
+	return nil
+}
+
+// Wait blocks until the worker has stopped.
+func (s *Service) Wait() {
+	s.wg.Wait()
+}
+
+// wakeWorker gives the worker a token to look for due rows.
+func (s *Service) wakeWorker() {
+	select {
+	case s.wake <- struct{}{}:
+	default: // the worker already has a token and will look again
+	}
+}
+
+// work delivers due rows until ctx is cancelled, then sleeps until the next
+// pending row is due or Emit stores one. A database error is retried with
+// backoff.
+func (s *Service) work(ctx context.Context) {
+	defer s.wg.Done()
+	backoff := pollRetryMin
+	for ctx.Err() == nil {
+		wait, err := s.deliverDue(ctx)
+		if err != nil {
+			if ctx.Err() != nil {
+				return
+			}
+			s.logger.ErrorContext(ctx, "deliver notifications failed; retrying", "error", err, "retry_in", backoff)
+			wait, backoff = backoff, min(2*backoff, pollRetryMax)
+		} else {
+			backoff = pollRetryMin
+		}
+		var timeout <-chan time.Time
+		var timer *time.Timer
+		if wait >= 0 {
+			timer = time.NewTimer(wait)
+			timeout = timer.C
+		}
+		select {
+		case <-s.wake:
+		case <-timeout:
+		case <-ctx.Done():
+		}
+		if timer != nil {
+			timer.Stop()
+		}
+	}
+}
+
+// deliverDue delivers every row due now and returns how long until the next
+// pending row is due, or -1 when none is pending.
+func (s *Service) deliverDue(ctx context.Context) (time.Duration, error) {
+	q := db.New(s.st.Writer)
+	for ctx.Err() == nil {
+		now := s.now()
+		lease := sql.NullString{String: timestamp(now.Add(leaseFor)), Valid: true}
+		row, err := q.ClaimDueNotification(ctx, db.ClaimDueNotificationParams{LeaseUntil: lease, Now: timestamp(now)})
+		if errors.Is(err, sql.ErrNoRows) {
+			break
+		}
+		if err != nil {
+			return 0, fmt.Errorf("notify: claim notification: %w", err)
+		}
+		s.deliver(ctx, row)
+	}
+	if err := ctx.Err(); err != nil {
+		return 0, err
+	}
+	next, err := db.New(s.st.Reader).NextNotificationDue(ctx)
+	if errors.Is(err, sql.ErrNoRows) {
+		return -1, nil
+	}
+	if err != nil {
+		return 0, fmt.Errorf("notify: read next due notification: %w", err)
+	}
+	due, err := parseTimestamp(next.String)
+	if err != nil {
+		return 0, fmt.Errorf("notify: next due notification: %w", err)
+	}
+	return max(due.Sub(s.now()), 0), nil
+}
+
+// deliver sends a claimed row through its channel's sender and records the
+// outcome. A row that cannot be sent at all, because its payload does not
+// decode or its channel has no sender, is failed at once: retrying cannot
+// help. A row 24 hours or more past its creation is failed without being
+// sent: it is too stale to deliver.
+func (s *Service) deliver(ctx context.Context, row db.NotificationOutbox) {
+	logger := s.logger.With("notification_id", row.ID, "event_key", row.EventKey, "channel", row.Channel)
+	var payload Payload
+	if err := json.Unmarshal([]byte(row.Payload), &payload); err != nil {
+		logger.ErrorContext(ctx, "notification failed", "reason", "payload does not decode")
+		s.giveUp(ctx, logger, row, 0, "payload does not decode")
+		return
+	}
+	s.mu.Lock()
+	sender, ok := s.senders[Channel(row.Channel)]
+	s.mu.Unlock()
+	if !ok {
+		reason := fmt.Sprintf("no sender registered for channel %s", row.Channel)
+		logger.ErrorContext(ctx, "notification failed", "reason", reason)
+		s.giveUp(ctx, logger, row, 0, reason)
+		return
+	}
+
+	created, err := parseTimestamp(row.CreatedAt)
+	if err != nil {
+		reason := fmt.Sprintf("created_at does not parse: %v", err)
+		logger.ErrorContext(ctx, "notification failed", "reason", reason)
+		s.giveUp(ctx, logger, row, 0, reason)
+		return
+	}
+	deadline := created.Add(giveUpAfter)
+	if !s.now().Before(deadline) {
+		reason := "not delivered within 24 hours of creation"
+		if row.LastError.Valid {
+			reason += "; last error: " + row.LastError.String
+		}
+		logger.ErrorContext(ctx, "notification failed; giving up", "reason", reason)
+		s.giveUp(ctx, logger, row, 0, reason)
+		return
+	}
+
+	attempt := row.Attempts + 1
+	sendCtx, cancel := context.WithTimeout(ctx, sendTimeout)
+	err = sender.Send(sendCtx, Message{
+		ID: row.ID, EventKey: row.EventKey, Channel: Channel(row.Channel), Payload: payload, Attempt: attempt,
+	})
+	cancel()
+	now := s.now()
+	if err == nil {
+		s.record(ctx, logger, "record notification sent", func(write context.Context) (int64, error) {
+			return db.New(s.st.Writer).MarkNotificationSent(write, db.MarkNotificationSentParams{
+				Now: timestamp(now), ID: row.ID, LeaseUntil: row.NextAttemptAt,
+			})
+		})
+		logger.InfoContext(ctx, "notification sent", "attempt", attempt)
+		return
+	}
+	if ctx.Err() != nil {
+		logger.WarnContext(context.WithoutCancel(ctx), "notification delivery interrupted by shutdown; it is retried when its lease ends")
+		return
+	}
+
+	if !now.Before(deadline) {
+		logger.ErrorContext(ctx, "notification failed; giving up", "attempt", attempt, "error", err)
+		s.giveUp(ctx, logger, row, 1, err.Error())
+		return
+	}
+	next := now.Add(retryDelay(attempt))
+	if next.After(deadline) {
+		next = deadline
+	}
+	logger.WarnContext(ctx, "notification delivery failed; retrying", "attempt", attempt, "error", err, "retry_at", timestamp(next))
+	s.record(ctx, logger, "record notification retry", func(write context.Context) (int64, error) {
+		return db.New(s.st.Writer).RetryNotification(write, db.RetryNotificationParams{
+			NextAttemptAt: sql.NullString{String: timestamp(next), Valid: true},
+			LastError:     sql.NullString{String: err.Error(), Valid: true},
+			Now:           timestamp(now),
+			ID:            row.ID,
+			LeaseUntil:    row.NextAttemptAt,
+		})
+	})
+}
+
+// giveUp marks a claimed row failed with reason. attempted is 1 when a
+// delivery was attempted, else 0.
+func (s *Service) giveUp(ctx context.Context, logger *slog.Logger, row db.NotificationOutbox, attempted int64, reason string) {
+	now := s.now()
+	s.record(ctx, logger, "record notification failed", func(write context.Context) (int64, error) {
+		return db.New(s.st.Writer).FailNotification(write, db.FailNotificationParams{
+			Attempted:  attempted,
+			LastError:  sql.NullString{String: reason, Valid: true},
+			Now:        timestamp(now),
+			ID:         row.ID,
+			LeaseUntil: row.NextAttemptAt,
+		})
+	})
+}
+
+// record makes a row's final write for one delivery, even when ctx has been
+// cancelled, bounded by jobs.FinalWriteTimeout. A write that fails, or that
+// finds the row no longer leased to this claim, is logged; the row keeps
+// its lease and is delivered again once the lease ends.
+func (s *Service) record(ctx context.Context, logger *slog.Logger, what string, write func(context.Context) (int64, error)) {
+	attempt, cancel := jobs.FinalWriteContext(ctx)
+	defer cancel()
+	n, err := write(attempt)
+	if err != nil {
+		logger.ErrorContext(context.WithoutCancel(ctx), what+" failed; it is delivered again when its lease ends", "error", err)
+		return
+	}
+	if n != 1 {
+		logger.ErrorContext(context.WithoutCancel(ctx), what, "error", "notification is no longer leased to this delivery")
+	}
+}
+
+// retryDelay is the wait after the attempt-th failed delivery: retryFirst,
+// doubling with each failure, at most retryMax.
+func retryDelay(attempt int64) time.Duration {
+	d := retryFirst
+	for i := int64(1); i < attempt && d < retryMax; i++ {
+		d *= 2
+	}
+	return min(d, retryMax)
+}

migrations/queries/notifications.sql

@@ -0,0 +1,58 @@
+-- name: GetNotificationPref :one
+SELECT * FROM notification_prefs
+WHERE event = ?;
+
+-- name: InsertNotification :execrows
+-- Stores a pending notification for one channel, due at once. A row already
+-- stored for the event key and channel, in any status, is left unchanged and
+-- no row is inserted.
+INSERT INTO notification_outbox (
+	id, event_key, channel, payload, status, attempts, next_attempt_at, created_at, updated_at
+) VALUES (
+	sqlc.arg(id), sqlc.arg(event_key), sqlc.arg(channel), sqlc.arg(payload), 'pending', 0, sqlc.arg(next_attempt_at), sqlc.arg(now), sqlc.arg(now)
+)
+ON CONFLICT (event_key, channel) DO NOTHING;
+
+-- name: ClaimDueNotification :one
+-- Leases the pending notification due longest ago by moving its
+-- next_attempt_at to lease_until, and returns it. A claimed row is not due
+-- again until the lease ends, so no other claim takes it while it is being
+-- delivered. Returns sql.ErrNoRows when none is due.
+UPDATE notification_outbox
+SET next_attempt_at = sqlc.arg(lease_until), updated_at = sqlc.arg(now)
+WHERE id = (
+	SELECT id FROM notification_outbox
+	WHERE status = 'pending' AND next_attempt_at <= sqlc.arg(now)
+	ORDER BY next_attempt_at, id
+	LIMIT 1
+)
+RETURNING *;
+
+-- name: MarkNotificationSent :execrows
+-- Records a delivered notification; updated_at is when it was sent. The
+-- lease must still be the claim's.
+UPDATE notification_outbox
+SET status = 'sent', attempts = attempts + 1, next_attempt_at = NULL, updated_at = sqlc.arg(now)
+WHERE id = sqlc.arg(id) AND status = 'pending' AND next_attempt_at = sqlc.arg(lease_until);
+
+-- name: RetryNotification :execrows
+-- Records a failed delivery and when to try again. The lease must still be
+-- the claim's.
+UPDATE notification_outbox
+SET attempts = attempts + 1, next_attempt_at = sqlc.arg(next_attempt_at), last_error = sqlc.arg(last_error), updated_at = sqlc.arg(now)
+WHERE id = sqlc.arg(id) AND status = 'pending' AND next_attempt_at = sqlc.arg(lease_until);
+
+-- name: FailNotification :execrows
+-- Gives up on a notification. attempts counts the delivery attempt that
+-- failed, if one was made. The lease must still be the claim's.
+UPDATE notification_outbox
+SET status = 'failed', attempts = attempts + sqlc.arg(attempted), next_attempt_at = NULL, last_error = sqlc.arg(last_error), updated_at = sqlc.arg(now)
+WHERE id = sqlc.arg(id) AND status = 'pending' AND next_attempt_at = sqlc.arg(lease_until);
+
+-- name: NextNotificationDue :one
+-- Returns when the next pending notification is due. Returns sql.ErrNoRows
+-- when none is pending.
+SELECT next_attempt_at FROM notification_outbox
+WHERE status = 'pending'
+ORDER BY next_attempt_at
+LIMIT 1;