diff --git a/cmd/server/main.go b/cmd/server/main.go index ed3f863..231e070 100644 --- a/cmd/server/main.go +++ b/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 - // database closes. - jobsCtx, stopJobs := context.WithCancel(ctx) + // worker runs. The job workers and the notification worker stop, and + // are waited for, before the database closes. + 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 diff --git a/internal/db/notifications.sql.go b/internal/db/notifications.sql.go new file mode 100644 index 0000000..3fea8f9 --- /dev/null +++ b/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() +} diff --git a/internal/db/querier.go b/internal/db/querier.go index 5c2ff44..5a3cfa9 100644 --- a/internal/db/querier.go +++ b/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. diff --git a/internal/notify/notify.go b/internal/notify/notify.go new file mode 100644 index 0000000..92a5906 --- /dev/null +++ b/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 +} diff --git a/internal/notify/notify_test.go b/internal/notify/notify_test.go new file mode 100644 index 0000000..16b3470 --- /dev/null +++ b/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) + } +} diff --git a/internal/notify/worker.go b/internal/notify/worker.go new file mode 100644 index 0000000..796e7c2 --- /dev/null +++ b/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) +} diff --git a/migrations/queries/notifications.sql b/migrations/queries/notifications.sql new file mode 100644 index 0000000..bda9d0b --- /dev/null +++ b/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;