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 TestEmitPushStoresOnlyAPushRow(t *testing.T) { ctx := context.Background() svc, st, _ := newTestService(t) setPref(t, st, PlanReady, true, true) const key = "plan_ready:test-1" if err := svc.EmitPush(ctx, planReady(key)); err != nil { t.Fatal(err) } r := onlyRow(t, st, key, ChannelPush) if r.Status != "pending" || r.NextAttemptAt.String != timestamp(start) { t.Errorf("row = %+v, want pending, due now", r) } select { case <-svc.wake: default: t.Error("EmitPush did not wake the worker") } // The same key again is ignored, as by Emit. if err := svc.EmitPush(ctx, planReady(key)); err != nil { t.Fatalf("duplicate EmitPush: %v", err) } if got := onlyRow(t, st, key, ChannelPush); got != r { t.Fatalf("duplicate EmitPush changed the row:\n got %+v\nwant %+v", got, r) } } func TestEmitPushRefusesWhenPushIsOff(t *testing.T) { svc, st, _ := newTestService(t) setPref(t, st, PlanReady, false, true) const key = "plan_ready:test-1" if err := svc.EmitPush(context.Background(), planReady(key)); !errors.Is(err, ErrPushOff) { t.Fatalf("EmitPush = %v, want ErrPushOff", err) } if rows := outbox(t, st, key); len(rows) != 0 { t.Fatalf("rows = %v, want none", rows) } } func TestEmitPushRejectsInvalidEvents(t *testing.T) { svc, _, _ := newTestService(t) ev := planReady("plan_ready:test-1") ev.URL = "https://evil.example/plan" if err := svc.EmitPush(context.Background(), ev); !errors.Is(err, ErrInvalidEvent) { t.Fatalf("EmitPush = %v, want ErrInvalidEvent", err) } } 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 TestUndeliverableFailsAtOnce(t *testing.T) { ctx := context.Background() svc, st, _ := newTestService(t) setPref(t, st, PlanReady, false, true) sender := &fakeSender{fail: fmt.Errorf("%w: chat refused the message (status 400)", ErrUndeliverable)} if err := svc.Register(ChannelChat, 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) } wait, err := svc.deliverDue(ctx) if err != nil { t.Fatal(err) } r := onlyRow(t, st, key, ChannelChat) if r.Status != "failed" || r.Attempts != 1 || r.LastError.String != sender.fail.Error() { t.Fatalf("row = %+v, want failed after one attempt with the sender's error", r) } if wait != -1 || len(sender.messages()) != 1 { t.Fatalf("wait = %v after %d deliveries, want nothing pending after one", wait, len(sender.messages())) } } 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) } } // TestEmitTxCommitsWithTheCallersTransaction: EmitTx's rows are stored only // when the caller's transaction commits. func TestEmitTxCommitsWithTheCallersTransaction(t *testing.T) { ctx := context.Background() svc, st, _ := newTestService(t) setPref(t, st, PlanReady, true, false) const key = "plan_ready:2026-10-12" rollback := errors.New("roll back") err := st.WithTx(ctx, func(tx *sql.Tx) error { if err := svc.EmitTx(ctx, tx, planReady(key)); err != nil { return err } return rollback }) if !errors.Is(err, rollback) { t.Fatalf("WithTx = %v", err) } if rows := outbox(t, st, key); len(rows) != 0 { t.Fatalf("rows after rollback = %v, want none", rows) } if err := st.WithTx(ctx, func(tx *sql.Tx) error { return svc.EmitTx(ctx, tx, planReady(key)) }); err != nil { t.Fatal(err) } if r := onlyRow(t, st, key, ChannelPush); r.Status != "pending" { t.Errorf("row after commit = %+v, want pending", r) } if err := st.WithTx(ctx, func(tx *sql.Tx) error { return svc.EmitTx(ctx, tx, Event{Type: PlanReady, Key: key}) }); !errors.Is(err, ErrInvalidEvent) { t.Errorf("EmitTx of an invalid event = %v, want ErrInvalidEvent", err) } }