nudge: add digest/batching mode for care nudges

When enabled, eligible nudges (sev ≤ ceiling) are queued in memory instead of sent immediately. Every 'window' duration (or when 'max_items' reached), the queue is flushed as a single digest notification with concatenated bodies. Config: digest: enabled: true # default false window: 30m # flush window (default 30m) max_items: 5 # flush at this count (default 5) severity_ceiling: 2 # max sev batched (default 2; sev3+ bypass) Changes: - config.go: add DigestConfig struct with defaults - tick.go: QueuedNudge type, digest queue/flush in TickLoop, shouldQueue/maybeFlush/flushDigest helpers - main.go: pass cfg.Digest to newTickLoop - tick_test.go: 6 new tests covering queue, flush, bypass, dedup
2026-07-05 13:07:28 +04:00
parent 5afff001c3
commit 354990fed0
5 changed files with 388 additions and 18 deletions
@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"

+	"github.com/kami/maven/internal/config"
 	"github.com/kami/maven/internal/delivery"
 	"github.com/kami/maven/internal/loop"
 	"github.com/kami/maven/internal/phraser"
@@ -33,7 +34,7 @@ func newTestStore(t *testing.T) *store.Store {
 	return st
 }

-func newTestTickLoop(t *testing.T, st *store.Store, sink delivery.Sink) *tickLoop {
+func newTestTickLoop(t *testing.T, st *store.Store, sink delivery.Sink, digestCfg *config.DigestConfig) *tickLoop {
 	t.Helper()
 	rules := loop.DefaultRules()
 	g := loop.NewGatherer(st, rules)
@@ -44,7 +45,7 @@ func newTestTickLoop(t *testing.T, st *store.Store, sink delivery.Sink) *tickLoo
 		Nudges:    st,
 		Reminders: st,
 	})
-	return newTickLoop(st, g, d, phraser.NewStub(), rules, time.Second, 5*time.Minute, 0)
+	return newTickLoop(st, g, d, phraser.NewStub(), rules, time.Second, 5*time.Minute, 0, digestCfg)
 }

 // refNow — fixed tick time so presence decay + since durations are deterministic.
@@ -69,7 +70,7 @@ func TestTickColdStoreSendsNothing(t *testing.T) {
 	st := newTestStore(t)
 	ctx := context.Background()
 	sink := &fakeSink{}
-	tl := newTestTickLoop(t, st, sink)
+	tl := newTestTickLoop(t, st, sink, nil)

 	tl.tick(ctx, refNow())

@@ -90,7 +91,7 @@ func TestTickWaterFiresWhenDueAndPresent(t *testing.T) {
 		t.Fatalf("seed water: %v", err)
 	}
 	sink := &fakeSink{}
-	tl := newTestTickLoop(t, st, sink)
+	tl := newTestTickLoop(t, st, sink, nil)

 	tl.tick(ctx, now)

@@ -131,7 +132,7 @@ func TestTickCooldownSuppressesSecondSend(t *testing.T) {
 		_ = err
 	}
 	sink := &fakeSink{}
-	tl := newTestTickLoop(t, st, sink)
+	tl := newTestTickLoop(t, st, sink, nil)

 	tl.tick(ctx, now)                 // fires
 	tl.tick(ctx, now.Add(time.Minute)) // still within 30m cooldown ⇒ suppressed
@@ -154,7 +155,7 @@ func TestTickReminderFiresOnceAndMarkedFired(t *testing.T) {
 		t.Fatalf("CreateReminder: %v", err)
 	}
 	sink := &fakeSink{}
-	tl := newTestTickLoop(t, st, sink)
+	tl := newTestTickLoop(t, st, sink, nil)

 	tl.tick(ctx, now)
 	if got, want := len(sink.sends), 1; got != want {
@@ -190,7 +191,7 @@ func TestTuneWritesFeedbackCooldownToStore(t *testing.T) {
 	now := refNow()

 	sink := &fakeSink{}
-	tl := newTestTickLoop(t, st, sink)
+	tl := newTestTickLoop(t, st, sink, nil)

 	// seed enough resolved `ignored` nudge outcomes to trip the tuner (above
 	// TuneMinOutcomes). all-ignored ⇒ factor 1.5 ⇒ base × 1.5; clamped to Max.
@@ -285,4 +286,227 @@ func TestTuneWritesFeedbackCooldownToStore(t *testing.T) {
 		t.Fatalf("gatherer used tuned base: CooldownUntil want %v, got %v",
 			wantUntil, got)
 	}
-}
+}
+
+// ----------------------------- digest / batching ----------------------------
+
+func TestDigestQueuesEligibleNudge(t *testing.T) {
+	// sev1 (water) with digest enabled → queued, not dispatched.
+	st := newTestStore(t)
+	ctx := context.Background()
+	now := refNow()
+	markPresent(t, st, ctx, now)
+	if _, err := st.SetValue(ctx, store.KindSelf, "water", "tap:water", map[string]int{"ml": 0}, now.Add(-4*time.Hour)); err != nil {
+		t.Fatalf("seed water: %v", err)
+	}
+	sink := &fakeSink{}
+	dc := &config.DigestConfig{
+		Enabled:         true,
+		Window:          config.Duration(30 * time.Minute),
+		MaxItems:        5,
+		SeverityCeiling: 2,
+	}
+	tl := newTestTickLoop(t, st, sink, dc)
+
+	tl.tick(ctx, now)
+
+	if len(sink.sends) != 0 {
+		t.Fatalf("digest: eligible nudge should not dispatch immediately, got %d sends", len(sink.sends))
+	}
+	if len(tl.digestQ) != 1 {
+		t.Fatalf("digestQ length = %d, want 1", len(tl.digestQ))
+	}
+	if tl.digestQ[0].Rule != "water" {
+		t.Errorf("queued rule = %q, want %q", tl.digestQ[0].Rule, "water")
+	}
+}
+
+func TestDigestFlushesAfterWindow(t *testing.T) {
+	// Queue a sev1 nudge, advance past the window, tick again → flush.
+	// Re-mark presence on the second tick so the flush dispatch has a
+	// non-away presence route (care nudges drop on away).
+	st := newTestStore(t)
+	ctx := context.Background()
+	now := refNow()
+	markPresent(t, st, ctx, now)
+	if _, err := st.SetValue(ctx, store.KindSelf, "water", "tap:water", map[string]int{"ml": 0}, now.Add(-4*time.Hour)); err != nil {
+		t.Fatalf("seed water: %v", err)
+	}
+	sink := &fakeSink{}
+	dc := &config.DigestConfig{
+		Enabled:         true,
+		Window:          config.Duration(30 * time.Minute),
+		MaxItems:        5,
+		SeverityCeiling: 2,
+	}
+	tl := newTestTickLoop(t, st, sink, dc)
+
+	// Tick 1: water queues, nothing dispatched.
+	tl.tick(ctx, now)
+	if len(sink.sends) != 0 {
+		t.Fatalf("tick 1: want 0 sends (queued), got %d", len(sink.sends))
+	}
+	if len(tl.digestQ) != 1 {
+		t.Fatalf("tick 1: digestQ = %d, want 1", len(tl.digestQ))
+	}
+
+	// Tick 2: window elapsed → flush. Re-mark presence to keep routing
+	// present (care nudge away → drop).
+	sink.sends = nil
+	later := now.Add(31 * time.Minute)
+	markPresent(t, st, ctx, later)
+	tl.tick(ctx, later)
+
+	if len(sink.sends) != 1 {
+		t.Fatalf("tick 2 (flush): sends = %d, want 1", len(sink.sends))
+	}
+	if sink.sends[0].RuleName != "digest" {
+		t.Errorf("flush rule = %q, want %q", sink.sends[0].RuleName, "digest")
+	}
+	if sink.sends[0].Body == "" {
+		t.Error("digest body must not be empty")
+	}
+	if len(tl.digestQ) != 0 {
+		t.Errorf("digestQ should be empty after flush, got %d", len(tl.digestQ))
+	}
+}
+
+func TestDigestFlushesAtMaxItems(t *testing.T) {
+	// Queue water via tick, then manually push a second item, then tick again
+	// so the before-candidate maybeFlush sees len >= MaxItems and flushes.
+	st := newTestStore(t)
+	ctx := context.Background()
+	now := refNow()
+	markPresent(t, st, ctx, now)
+
+	if _, err := st.SetValue(ctx, store.KindSelf, "water", "tap:water", map[string]int{"ml": 0}, now.Add(-4*time.Hour)); err != nil {
+		t.Fatalf("seed water: %v", err)
+	}
+
+	sink := &fakeSink{}
+	dc := &config.DigestConfig{
+		Enabled:         true,
+		Window:          config.Duration(30 * time.Minute),
+		MaxItems:        2,
+		SeverityCeiling: 2,
+	}
+	tl := newTestTickLoop(t, st, sink, dc)
+
+	// Tick 1: water queues (1 item, below MaxItems).
+	tl.tick(ctx, now)
+	if len(tl.digestQ) != 1 {
+		t.Fatalf("tick 1: digestQ = %d, want 1", len(tl.digestQ))
+	}
+	if len(sink.sends) != 0 {
+		t.Fatalf("tick 1: sends = %d, want 0", len(sink.sends))
+	}
+
+	// Manually push a second item to hit MaxItems.
+	tl.digestQ = append(tl.digestQ, QueuedNudge{
+		Rule: "meal", Severity: 1, Body: "eat something", Key: "meal", QueuedAt: now.Add(time.Second),
+	})
+
+	// Tick 2: after-candidate maybeFlush sees len=2 >= MaxItems=2 → flush.
+	sink.sends = nil
+	tl.tick(ctx, now.Add(2*time.Second))
+
+	if len(sink.sends) != 1 {
+		t.Fatalf("after flush tick: sends = %d, want 1", len(sink.sends))
+	}
+	if sink.sends[0].RuleName != "digest" {
+		t.Errorf("flush rule = %q, want %q", sink.sends[0].RuleName, "digest")
+	}
+	if len(tl.digestQ) != 0 {
+		t.Errorf("digestQ should be empty after flush, got %d", len(tl.digestQ))
+	}
+}
+
+func TestDigestSev4BypassesQueue(t *testing.T) {
+	// Sev4 (service_down) with digest enabled → dispatches immediately.
+	st := newTestStore(t)
+	ctx := context.Background()
+	now := refNow()
+	markPresent(t, st, ctx, now)
+	if _, err := st.SetValue(ctx, store.KindSelf, "service_down", "poll:uptimekuma", "down", now); err != nil {
+		t.Fatalf("seed service_down: %v", err)
+	}
+	sink := &fakeSink{}
+	dc := &config.DigestConfig{
+		Enabled:         true,
+		Window:          config.Duration(30 * time.Minute),
+		MaxItems:        5,
+		SeverityCeiling: 2,
+	}
+	tl := newTestTickLoop(t, st, sink, dc)
+
+	tl.tick(ctx, now)
+
+	if len(sink.sends) == 0 {
+		t.Fatal("sev4 nudge must dispatch immediately, bypassing digest queue")
+	}
+	if len(tl.digestQ) != 0 {
+		t.Errorf("digestQ should be empty (sev4 bypassed), got %d", len(tl.digestQ))
+	}
+}
+
+func TestDigestDisabledSendsImmediately(t *testing.T) {
+	// Digest disabled (nil config) → sev1 dispatches immediately.
+	st := newTestStore(t)
+	ctx := context.Background()
+	now := refNow()
+	markPresent(t, st, ctx, now)
+	if _, err := st.SetValue(ctx, store.KindSelf, "water", "tap:water", map[string]int{"ml": 0}, now.Add(-4*time.Hour)); err != nil {
+		t.Fatalf("seed water: %v", err)
+	}
+	sink := &fakeSink{}
+	tl := newTestTickLoop(t, st, sink, nil)
+
+	tl.tick(ctx, now)
+
+	if len(sink.sends) != 1 {
+		t.Fatalf("digest disabled: sends = %d, want 1", len(sink.sends))
+	}
+	if sink.sends[0].RuleName != "water" {
+		t.Errorf("send rule = %q, want %q", sink.sends[0].RuleName, "water")
+	}
+}
+
+func TestDigestDeduplicatesByRule(t *testing.T) {
+	// Same rule (water) fires in two ticks; only one copy in the queue.
+	st := newTestStore(t)
+	ctx := context.Background()
+	now := refNow()
+	markPresent(t, st, ctx, now)
+	if _, err := st.SetValue(ctx, store.KindSelf, "water", "tap:water", map[string]int{"ml": 0}, now.Add(-4*time.Hour)); err != nil {
+		t.Fatalf("seed water: %v", err)
+	}
+	sink := &fakeSink{}
+	dc := &config.DigestConfig{
+		Enabled:         true,
+		Window:          config.Duration(30 * time.Minute),
+		MaxItems:        5,
+		SeverityCeiling: 2,
+	}
+	tl := newTestTickLoop(t, st, sink, dc)
+
+	// Tick 1: water queues.
+	tl.tick(ctx, now)
+	if len(tl.digestQ) != 1 {
+		t.Fatalf("tick 1: digestQ = %d, want 1", len(tl.digestQ))
+	}
+
+	// Tick 2: water still in cooldown → gate suppresses, but if cooldown
+	// weren't active the dedup check would prevent a second copy. We verify
+	// by making a second direct call to queueNudge with the same rule name.
+	// We also tick again with a different time to see if water fires again
+	// through the normal path — but cooldown should suppress it. Instead,
+	// directly verify dedup by calling queueNudge with a synthetic candidate.
+	tl.queueNudge(ctx, &loop.Candidate{
+		Rule:     loop.Rule{Name: "water", Severity: loop.Sev1},
+		Severity: loop.Sev1,
+	}, loop.State{}, now.Add(time.Minute))
+
+	if len(tl.digestQ) != 1 {
+		t.Fatalf("after duplicate queue attempt: digestQ = %d, want 1 (dedup)", len(tl.digestQ))
+	}
+}