kami
29f23e3715
Closes the audit finding: a crash between 'sink accepted it' and 'we recorded that' caused duplicate sends on the next tick with no trace. BeginDeliveryAttempt now runs before Send, CompleteDeliveryAttempt after — a stale 'pending' row found at startup reconciles to 'unknown' (never silently resent, never silently dropped, same rule as the Hexis execution engine's timeout handling). Wired into DispatchNudge, DispatchReminder, and RepeatUnacked; ReconcileStaleDeliveryAttempts runs once at mavend startup before the tick loop resumes. 9 new dispatcher tests cover begin-before-send ordering, failed-send completion, the reminder path, and begin-failure not blocking send. Vikunja #270. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ghELqYhZNLub2TXGMazqA
787 lines
26 KiB
Go
787 lines
26 KiB
Go
package delivery
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/kami/maven/internal/loop"
|
|
"github.com/kami/maven/internal/store"
|
|
)
|
|
|
|
func refNow() time.Time { return time.Date(2026, 6, 30, 12, 0, 0, 0, time.UTC) }
|
|
|
|
func sevRule(name string, sev loop.Severity) loop.Rule {
|
|
return loop.Rule{Name: name, Severity: sev}
|
|
}
|
|
|
|
func candidate(name string, sev loop.Severity, presence store.Bucket) loop.Candidate {
|
|
return loop.Candidate{
|
|
Rule: sevRule(name, sev),
|
|
Severity: sev,
|
|
State: loop.State{Now: refNow(), Presence: presence},
|
|
}
|
|
}
|
|
|
|
// --- fakes ---
|
|
|
|
type fakeSink struct {
|
|
sends []Sendable
|
|
err error
|
|
}
|
|
|
|
func (f *fakeSink) Send(_ context.Context, s Sendable) error {
|
|
if f.err != nil {
|
|
return f.err
|
|
}
|
|
f.sends = append(f.sends, s)
|
|
return nil
|
|
}
|
|
|
|
type fakeNudgeRecorder struct {
|
|
rows []nudgeRow
|
|
id int64
|
|
}
|
|
|
|
type nudgeRow struct {
|
|
rule, channel, message string
|
|
ts time.Time
|
|
}
|
|
|
|
func (f *fakeNudgeRecorder) RecordNudge(_ context.Context, rule, channel, message string, ts time.Time) (int64, error) {
|
|
f.id++
|
|
f.rows = append(f.rows, nudgeRow{rule, channel, message, ts})
|
|
return f.id, nil
|
|
}
|
|
|
|
type fakeReminderCompleter struct {
|
|
marked []struct {
|
|
id int64
|
|
status string
|
|
}
|
|
rescheduled []int64
|
|
err error
|
|
}
|
|
|
|
func (f *fakeReminderCompleter) MarkReminder(_ context.Context, id int64, status string) error {
|
|
if f.err != nil {
|
|
return f.err
|
|
}
|
|
f.marked = append(f.marked, struct {
|
|
id int64
|
|
status string
|
|
}{id, status})
|
|
return nil
|
|
}
|
|
|
|
func (f *fakeReminderCompleter) RescheduleReminder(_ context.Context, id int64, _ time.Time) error {
|
|
if f.err != nil {
|
|
return f.err
|
|
}
|
|
f.rescheduled = append(f.rescheduled, id)
|
|
return nil
|
|
}
|
|
|
|
type fakeAck struct {
|
|
acked map[string]bool
|
|
lastSent map[string]time.Time
|
|
}
|
|
|
|
func newFakeAck() *fakeAck {
|
|
return &fakeAck{acked: make(map[string]bool), lastSent: make(map[string]time.Time)}
|
|
}
|
|
|
|
func (f *fakeAck) WasAcked(_ context.Context, key string) (bool, error) {
|
|
return f.acked[key], nil
|
|
}
|
|
func (f *fakeAck) MarkSent(_ context.Context, key string, ts time.Time) error {
|
|
f.lastSent[key] = ts
|
|
return nil
|
|
}
|
|
func (f *fakeAck) LastSent(_ context.Context, key string) (time.Time, error) {
|
|
return f.lastSent[key], nil
|
|
}
|
|
func (f *fakeAck) MarkAcked(_ context.Context, key string) error {
|
|
f.acked[key] = true
|
|
return nil
|
|
}
|
|
|
|
// ----------------------------- routing table --------------------------------
|
|
|
|
func TestChannelsForCarePresent(t *testing.T) {
|
|
got := ChannelsFor(loop.Sev1, store.Present)
|
|
if len(got) != 1 || got[0] != ChannelVoice {
|
|
t.Fatalf("sev1 present: want [voice], got %v", got)
|
|
}
|
|
}
|
|
|
|
func TestChannelsForCareAwayDrops(t *testing.T) {
|
|
// sev ≤ 2 drops on away — "a missed water nudge is noise."
|
|
got := ChannelsFor(loop.Sev2, store.Away)
|
|
if len(got) != 1 || got[0] != ChannelDrop {
|
|
t.Fatalf("sev2 away: want [drop], got %v", got)
|
|
}
|
|
}
|
|
|
|
func TestChannelsForOpsSoftAwayNtfyOnce(t *testing.T) {
|
|
got := ChannelsFor(loop.Sev3, store.Away)
|
|
if len(got) != 1 || got[0] != ChannelNtfy {
|
|
t.Fatalf("sev3 away: want [ntfy], got %v", got)
|
|
}
|
|
}
|
|
|
|
func TestChannelsForOpsHardPresentVoiceAndNtfy(t *testing.T) {
|
|
got := ChannelsFor(loop.Sev4, store.Present)
|
|
if len(got) != 2 || got[0] != ChannelVoice || got[1] != ChannelNtfy {
|
|
t.Fatalf("sev4 present: want [voice ntfy], got %v", got)
|
|
}
|
|
}
|
|
|
|
func TestChannelsForOpsHardAwayTelegram(t *testing.T) {
|
|
got := ChannelsFor(loop.Sev4, store.Away)
|
|
if len(got) != 1 || got[0] != ChannelTelegram {
|
|
t.Fatalf("sev4 away: want [telegram], got %v", got)
|
|
}
|
|
}
|
|
|
|
func TestChannelsForReminderPresentVoice(t *testing.T) {
|
|
got := ChannelsForReminder(store.Present)
|
|
if len(got) != 1 || got[0] != ChannelVoice {
|
|
t.Fatalf("reminder present: want [voice], got %v", got)
|
|
}
|
|
}
|
|
|
|
func TestChannelsForReminderAwayNtfy(t *testing.T) {
|
|
got := ChannelsForReminder(store.Away)
|
|
if len(got) != 1 || got[0] != ChannelNtfy {
|
|
t.Fatalf("reminder away: want [ntfy], got %v", got)
|
|
}
|
|
}
|
|
|
|
// ----------------------------- dispatcher: nudges ---------------------------
|
|
|
|
func TestDispatchNudgeCarePresentSendsVoice(t *testing.T) {
|
|
voice := &fakeSink{}
|
|
ntfy := &fakeSink{}
|
|
rec := &fakeNudgeRecorder{}
|
|
d := NewDispatcher(Config{Voice: voice, Ntfy: ntfy, Nudges: rec})
|
|
|
|
out, err := d.DispatchNudge(context.Background(), PhrasedNudge{
|
|
Candidate: candidate("water", loop.Sev1, store.Present),
|
|
Body: "you haven't had water in 4h",
|
|
Summary: "water",
|
|
}, refNow())
|
|
if err != nil {
|
|
t.Fatalf("dispatch: %v", err)
|
|
}
|
|
if len(out) != 1 || out[0].Sendable.Channel != ChannelVoice {
|
|
t.Fatalf("want 1 voice dispatch, got %+v", out)
|
|
}
|
|
if len(voice.sends) != 1 || voice.sends[0].Body != "you haven't had water in 4h" {
|
|
t.Fatalf("voice send: %+v", voice.sends)
|
|
}
|
|
if len(ntfy.sends) != 0 {
|
|
t.Fatalf("ntfy should not fire for sev1 present, got %+v", ntfy.sends)
|
|
}
|
|
if len(rec.rows) != 1 || rec.rows[0].rule != "water" || rec.rows[0].channel != "voice" {
|
|
t.Fatalf("nudge record: %+v", rec.rows)
|
|
}
|
|
}
|
|
|
|
func TestDispatchNudgeCareAwayDropsNoSendNoRecord(t *testing.T) {
|
|
voice := &fakeSink{}
|
|
rec := &fakeNudgeRecorder{}
|
|
d := NewDispatcher(Config{Voice: voice, Nudges: rec})
|
|
|
|
out, err := d.DispatchNudge(context.Background(), PhrasedNudge{
|
|
Candidate: candidate("water", loop.Sev1, store.Away),
|
|
Body: "you haven't had water in 4h",
|
|
Summary: "water",
|
|
}, refNow())
|
|
if err != nil {
|
|
t.Fatalf("dispatch: %v", err)
|
|
}
|
|
if len(out) != 0 {
|
|
t.Fatalf("care away: want 0 dispatches, got %+v", out)
|
|
}
|
|
if len(voice.sends) != 0 || len(rec.rows) != 0 {
|
|
t.Fatalf("drop = no send, no record; sends=%v rows=%v", voice.sends, rec.rows)
|
|
}
|
|
}
|
|
|
|
func TestDispatchNudgeOpsHardPresentSendsVoiceAndNtfy(t *testing.T) {
|
|
voice := &fakeSink{}
|
|
ntfy := &fakeSink{}
|
|
rec := &fakeNudgeRecorder{}
|
|
d := NewDispatcher(Config{Voice: voice, Ntfy: ntfy, Nudges: rec})
|
|
|
|
out, err := d.DispatchNudge(context.Background(), PhrasedNudge{
|
|
Candidate: candidate("service_down", loop.Sev4, store.Present),
|
|
Body: "the backup service on homesrv is down",
|
|
Summary: "backup down on homesrv",
|
|
}, refNow())
|
|
if err != nil {
|
|
t.Fatalf("dispatch: %v", err)
|
|
}
|
|
if len(out) != 2 {
|
|
t.Fatalf("sev4 present: want 2 dispatches, got %d", len(out))
|
|
}
|
|
if len(voice.sends) != 1 || len(ntfy.sends) != 1 {
|
|
t.Fatalf("want 1 voice + 1 ntfy, got voice=%d ntfy=%d", len(voice.sends), len(ntfy.sends))
|
|
}
|
|
}
|
|
|
|
func TestDispatchNudgeOpsHardAwayTelegramRepeatUntilAck(t *testing.T) {
|
|
telegram := &fakeSink{}
|
|
rec := &fakeNudgeRecorder{}
|
|
ack := newFakeAck()
|
|
d := NewDispatcher(Config{Telegram: telegram, Ack: ack, Nudges: rec})
|
|
|
|
out, err := d.DispatchNudge(context.Background(), PhrasedNudge{
|
|
Candidate: candidate("service_down", loop.Sev4, store.Away),
|
|
Body: "the backup service on homesrv is down",
|
|
Summary: "backup down on homesrv",
|
|
}, refNow())
|
|
if err != nil {
|
|
t.Fatalf("dispatch: %v", err)
|
|
}
|
|
if len(out) != 1 || out[0].Sendable.Channel != ChannelTelegram {
|
|
t.Fatalf("sev4 away: want 1 telegram, got %+v", out)
|
|
}
|
|
if !out[0].Sendable.RepeatUntilAck {
|
|
t.Fatalf("sev4 telegram away: want RepeatUntilAck=true")
|
|
}
|
|
if len(telegram.sends) != 1 {
|
|
t.Fatalf("want 1 telegram send, got %d", len(telegram.sends))
|
|
}
|
|
// ack tracker should have the initial MarkSent
|
|
last, _ := ack.LastSent(context.Background(), "service_down")
|
|
if !last.Equal(refNow()) {
|
|
t.Fatalf("ack MarkSent: want %v, got %v", refNow(), last)
|
|
}
|
|
}
|
|
|
|
func TestDispatchNudgeMinimalBodyForAwayChannels(t *testing.T) {
|
|
// away channels get Summary, not Body — the "minimal body" / no-shoulder-
|
|
// surf-exfil rule. the nudge record stores the summary too.
|
|
ntfy := &fakeSink{}
|
|
rec := &fakeNudgeRecorder{}
|
|
d := NewDispatcher(Config{Ntfy: ntfy, Nudges: rec})
|
|
|
|
_, err := d.DispatchNudge(context.Background(), PhrasedNudge{
|
|
Candidate: candidate("cert_expiring", loop.Sev3, store.Away),
|
|
Body: "the tls cert for homesrv.kami.lan expires in 3 days — renew via acme.sh on the reverse proxy",
|
|
Summary: "cert expiring soon",
|
|
}, refNow())
|
|
if err != nil {
|
|
t.Fatalf("dispatch: %v", err)
|
|
}
|
|
if ntfy.sends[0].Summary != "cert expiring soon" {
|
|
t.Fatalf("ntfy summary: want 'cert expiring soon', got %q", ntfy.sends[0].Summary)
|
|
}
|
|
if rec.rows[0].message != "cert expiring soon" {
|
|
t.Fatalf("recorded message should be summary, got %q", rec.rows[0].message)
|
|
}
|
|
}
|
|
|
|
func TestDispatchNudgeSendErrorStopsAndReturnsPartial(t *testing.T) {
|
|
// sev4 present → voice + ntfy. voice fails → ntfy never tried, partial
|
|
// returned. a failed send doesn't pollute the feedback loop (no record
|
|
// for the failed channel).
|
|
voice := &fakeSink{err: errors.New("audio device gone")}
|
|
ntfy := &fakeSink{}
|
|
rec := &fakeNudgeRecorder{}
|
|
d := NewDispatcher(Config{Voice: voice, Ntfy: ntfy, Nudges: rec})
|
|
|
|
out, err := d.DispatchNudge(context.Background(), PhrasedNudge{
|
|
Candidate: candidate("service_down", loop.Sev4, store.Present),
|
|
Body: "down", Summary: "down",
|
|
}, refNow())
|
|
if err == nil {
|
|
t.Fatalf("want send error, got nil")
|
|
}
|
|
if len(out) != 0 {
|
|
t.Fatalf("voice failed first → 0 dispatches, got %d", len(out))
|
|
}
|
|
if len(rec.rows) != 0 {
|
|
t.Fatalf("failed send must not record a nudge, got %d rows", len(rec.rows))
|
|
}
|
|
}
|
|
|
|
func TestDispatchNudgeNilSinkSkipsSilently(t *testing.T) {
|
|
// ntfy not wired; sev3 away routes to ntfy → skipped, no error.
|
|
rec := &fakeNudgeRecorder{}
|
|
d := NewDispatcher(Config{Nudges: rec})
|
|
|
|
out, err := d.DispatchNudge(context.Background(), PhrasedNudge{
|
|
Candidate: candidate("cert_expiring", loop.Sev3, store.Away),
|
|
Body: "cert", Summary: "cert",
|
|
}, refNow())
|
|
if err != nil {
|
|
t.Fatalf("nil sink should skip silently, got %v", err)
|
|
}
|
|
if len(out) != 0 {
|
|
t.Fatalf("nil ntfy → 0 dispatches, got %d", len(out))
|
|
}
|
|
}
|
|
|
|
// ----------------------------- dispatcher: reminders ------------------------
|
|
|
|
func TestDispatchReminderPresentVoice(t *testing.T) {
|
|
voice := &fakeSink{}
|
|
rc := &fakeReminderCompleter{}
|
|
d := NewDispatcher(Config{Voice: voice, Reminders: rc})
|
|
|
|
rd := loop.ReminderDecision{
|
|
Reminder: store.Reminder{ID: 42, Payload: `"wake me"`, Status: "pending"},
|
|
State: loop.State{Now: refNow(), Presence: store.Present},
|
|
}
|
|
out, err := d.DispatchReminder(context.Background(), PhrasedReminder{
|
|
Decision: rd, Body: "wake up", Summary: "wake up",
|
|
}, refNow())
|
|
if err != nil {
|
|
t.Fatalf("dispatch: %v", err)
|
|
}
|
|
if len(out) != 1 || out[0].Sendable.Channel != ChannelVoice {
|
|
t.Fatalf("want 1 voice, got %+v", out)
|
|
}
|
|
if len(rc.marked) != 1 || rc.marked[0].id != 42 || rc.marked[0].status != "fired" {
|
|
t.Fatalf("reminder not marked fired: %+v", rc.marked)
|
|
}
|
|
}
|
|
|
|
func TestDispatchReminderAwayNtfy(t *testing.T) {
|
|
ntfy := &fakeSink{}
|
|
rc := &fakeReminderCompleter{}
|
|
d := NewDispatcher(Config{Ntfy: ntfy, Reminders: rc})
|
|
|
|
rd := loop.ReminderDecision{
|
|
Reminder: store.Reminder{ID: 7, Status: "pending"},
|
|
State: loop.State{Now: refNow(), Presence: store.Away},
|
|
}
|
|
out, err := d.DispatchReminder(context.Background(), PhrasedReminder{
|
|
Decision: rd, Body: "full wake up message", Summary: "wake up",
|
|
}, refNow())
|
|
if err != nil {
|
|
t.Fatalf("dispatch: %v", err)
|
|
}
|
|
if len(out) != 1 || out[0].Sendable.Channel != ChannelNtfy {
|
|
t.Fatalf("want 1 ntfy, got %+v", out)
|
|
}
|
|
// away channel gets summary, not body
|
|
if ntfy.sends[0].Summary != "wake up" {
|
|
t.Fatalf("ntfy summary: want 'wake up', got %q", ntfy.sends[0].Summary)
|
|
}
|
|
if len(rc.marked) != 1 || rc.marked[0].status != "fired" {
|
|
t.Fatalf("reminder not marked fired: %+v", rc.marked)
|
|
}
|
|
}
|
|
|
|
func TestDispatchReminderFailedSendNotMarkedFired(t *testing.T) {
|
|
// a failed send must not mark the reminder fired — it stays pending for
|
|
// the next tick to re-deliver. same instinct as "record after success."
|
|
voice := &fakeSink{err: errors.New("no audio")}
|
|
rc := &fakeReminderCompleter{}
|
|
d := NewDispatcher(Config{Voice: voice, Reminders: rc})
|
|
|
|
rd := loop.ReminderDecision{
|
|
Reminder: store.Reminder{ID: 1, Status: "pending"},
|
|
State: loop.State{Now: refNow(), Presence: store.Present},
|
|
}
|
|
_, err := d.DispatchReminder(context.Background(), PhrasedReminder{
|
|
Decision: rd, Body: "wake", Summary: "wake",
|
|
}, refNow())
|
|
if err == nil {
|
|
t.Fatalf("want send error")
|
|
}
|
|
if len(rc.marked) != 0 {
|
|
t.Fatalf("failed send must not mark fired, got %+v", rc.marked)
|
|
}
|
|
}
|
|
|
|
// ------------------ voice no-session → away-channel reroute -----------------
|
|
//
|
|
// When the routing table picks voice (presence=present) but no live session
|
|
// exists at push time, the presence guess was wrong. The dispatcher must
|
|
// reroute through the AWAY table (§ away-channel fallthrough), not silently
|
|
// drop or fall to the wrong channel.
|
|
|
|
func TestDispatchNudgeVoiceNoSessionSev3RoutesNtfy(t *testing.T) {
|
|
// present sev3 → [voice]. voice has no session → away sev3 = ntfy.
|
|
voice := &fakeSink{err: ErrVoiceNoSession}
|
|
ntfy := &fakeSink{}
|
|
telegram := &fakeSink{}
|
|
rec := &fakeNudgeRecorder{}
|
|
d := NewDispatcher(Config{Voice: voice, Ntfy: ntfy, Telegram: telegram, Nudges: rec})
|
|
|
|
out, err := d.DispatchNudge(context.Background(), PhrasedNudge{
|
|
Candidate: candidate("cert_expiring", loop.Sev3, store.Present),
|
|
Body: "cert expiring", Summary: "cert expiring",
|
|
}, refNow())
|
|
if err != nil {
|
|
t.Fatalf("dispatch: %v", err)
|
|
}
|
|
if len(ntfy.sends) != 1 {
|
|
t.Fatalf("sev3 voice-no-session: want 1 ntfy send, got %d", len(ntfy.sends))
|
|
}
|
|
if len(telegram.sends) != 0 {
|
|
t.Fatalf("sev3 must not hit telegram, got %d", len(telegram.sends))
|
|
}
|
|
if len(out) != 1 || out[0].Sendable.Channel != ChannelNtfy {
|
|
t.Fatalf("want 1 ntfy dispatch, got %+v", out)
|
|
}
|
|
}
|
|
|
|
func TestDispatchNudgeVoiceNoSessionSev4RoutesTelegramRepeatUntilAck(t *testing.T) {
|
|
// present sev4 → [voice, ntfy]. voice has no session → away sev4 =
|
|
// telegram-repeat-til-ack (NOT the present-list ntfy remainder).
|
|
voice := &fakeSink{err: ErrVoiceNoSession}
|
|
ntfy := &fakeSink{}
|
|
telegram := &fakeSink{}
|
|
ack := newFakeAck()
|
|
rec := &fakeNudgeRecorder{}
|
|
d := NewDispatcher(Config{Voice: voice, Ntfy: ntfy, Telegram: telegram, Ack: ack, Nudges: rec})
|
|
|
|
out, err := d.DispatchNudge(context.Background(), PhrasedNudge{
|
|
Candidate: candidate("service_down", loop.Sev4, store.Present),
|
|
Body: "backup down", Summary: "backup down",
|
|
}, refNow())
|
|
if err != nil {
|
|
t.Fatalf("dispatch: %v", err)
|
|
}
|
|
if len(telegram.sends) != 1 {
|
|
t.Fatalf("sev4 voice-no-session: want 1 telegram send, got %d", len(telegram.sends))
|
|
}
|
|
if len(ntfy.sends) != 0 {
|
|
t.Fatalf("sev4 away reroute must not fall to ntfy, got %d", len(ntfy.sends))
|
|
}
|
|
if len(out) != 1 || out[0].Sendable.Channel != ChannelTelegram || !out[0].Sendable.RepeatUntilAck {
|
|
t.Fatalf("want 1 telegram RepeatUntilAck dispatch, got %+v", out)
|
|
}
|
|
if _, ok := ack.lastSent["service_down"]; !ok {
|
|
t.Fatalf("repeat-til-ack reroute must MarkSent in the ack tracker")
|
|
}
|
|
}
|
|
|
|
func TestDispatchNudgeVoiceNoSessionSev2Drops(t *testing.T) {
|
|
// present sev2 → [voice]. voice has no session → away sev2 = drop.
|
|
voice := &fakeSink{err: ErrVoiceNoSession}
|
|
ntfy := &fakeSink{}
|
|
telegram := &fakeSink{}
|
|
rec := &fakeNudgeRecorder{}
|
|
d := NewDispatcher(Config{Voice: voice, Ntfy: ntfy, Telegram: telegram, Nudges: rec})
|
|
|
|
out, err := d.DispatchNudge(context.Background(), PhrasedNudge{
|
|
Candidate: candidate("water", loop.Sev2, store.Present),
|
|
Body: "drink water", Summary: "drink water",
|
|
}, refNow())
|
|
if err != nil {
|
|
t.Fatalf("dispatch: %v", err)
|
|
}
|
|
if len(ntfy.sends) != 0 || len(telegram.sends) != 0 || len(out) != 0 {
|
|
t.Fatalf("sev2 voice-no-session must drop silently, got ntfy=%d telegram=%d out=%d",
|
|
len(ntfy.sends), len(telegram.sends), len(out))
|
|
}
|
|
}
|
|
|
|
func TestDispatchReminderVoiceNoSessionRoutesNtfy(t *testing.T) {
|
|
// present reminder → [voice]. voice has no session → away = ntfy.
|
|
voice := &fakeSink{err: ErrVoiceNoSession}
|
|
ntfy := &fakeSink{}
|
|
rc := &fakeReminderCompleter{}
|
|
d := NewDispatcher(Config{Voice: voice, Ntfy: ntfy, Reminders: rc})
|
|
|
|
rd := loop.ReminderDecision{
|
|
Reminder: store.Reminder{ID: 99, Status: "pending"},
|
|
State: loop.State{Now: refNow(), Presence: store.Present},
|
|
}
|
|
out, err := d.DispatchReminder(context.Background(), PhrasedReminder{
|
|
Decision: rd, Body: "wake up", Summary: "wake up",
|
|
}, refNow())
|
|
if err != nil {
|
|
t.Fatalf("dispatch: %v", err)
|
|
}
|
|
if len(ntfy.sends) != 1 || len(out) != 1 || out[0].Sendable.Channel != ChannelNtfy {
|
|
t.Fatalf("reminder voice-no-session: want 1 ntfy, got ntfy=%d out=%+v", len(ntfy.sends), out)
|
|
}
|
|
if len(rc.marked) != 1 || rc.marked[0].status != "fired" {
|
|
t.Fatalf("rerouted reminder must be marked fired: %+v", rc.marked)
|
|
}
|
|
}
|
|
|
|
// ----------------------------- repeat-til-ack -------------------------------
|
|
|
|
func TestShouldRepeat(t *testing.T) {
|
|
now := refNow()
|
|
cases := []struct {
|
|
name string
|
|
lastSent time.Time
|
|
acked bool
|
|
now time.Time
|
|
interval time.Duration
|
|
want bool
|
|
}{
|
|
{"never sent, not acked", time.Time{}, false, now, 5 * time.Minute, true},
|
|
{"acked → stop", now.Add(-1 * time.Minute), true, now, 5 * time.Minute, false},
|
|
{"interval not elapsed → wait", now.Add(-1 * time.Minute), false, now, 5 * time.Minute, false},
|
|
{"interval elapsed → repeat", now.Add(-6 * time.Minute), false, now, 5 * time.Minute, true},
|
|
{"exactly interval → repeat", now.Add(-5 * time.Minute), false, now, 5 * time.Minute, true},
|
|
}
|
|
for _, c := range cases {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
got := ShouldRepeat(c.lastSent, c.acked, c.now, c.interval)
|
|
if got != c.want {
|
|
t.Fatalf("ShouldRepeat: want %v, got %v", c.want, got)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestRepeatUnackedReSendsAfterInterval(t *testing.T) {
|
|
telegram := &fakeSink{}
|
|
ack := newFakeAck()
|
|
d := NewDispatcher(Config{Telegram: telegram, Ack: ack})
|
|
|
|
// initial send was 6min ago, interval 5min → should repeat.
|
|
initial := refNow().Add(-6 * time.Minute)
|
|
_ = ack.MarkSent(context.Background(), "service_down", initial)
|
|
|
|
out, err := d.RepeatUnacked(context.Background(), []string{"service_down"}, refNow(), 5*time.Minute, "down", "backup down")
|
|
if err != nil {
|
|
t.Fatalf("repeat: %v", err)
|
|
}
|
|
if len(out) != 1 || out[0].Sendable.RuleName != "service_down" {
|
|
t.Fatalf("want 1 repeat for service_down, got %+v", out)
|
|
}
|
|
// last-sent updated to now
|
|
last, _ := ack.LastSent(context.Background(), "service_down")
|
|
if !last.Equal(refNow()) {
|
|
t.Fatalf("last-sent should update to now, got %v", last)
|
|
}
|
|
}
|
|
|
|
func TestRepeatUnackedSkipsAcked(t *testing.T) {
|
|
telegram := &fakeSink{}
|
|
ack := newFakeAck()
|
|
d := NewDispatcher(Config{Telegram: telegram, Ack: ack})
|
|
|
|
_ = ack.MarkSent(context.Background(), "service_down", refNow().Add(-10*time.Minute))
|
|
_ = ack.MarkAcked(context.Background(), "service_down")
|
|
|
|
out, err := d.RepeatUnacked(context.Background(), []string{"service_down"}, refNow(), 5*time.Minute, "down", "down")
|
|
if err != nil {
|
|
t.Fatalf("repeat: %v", err)
|
|
}
|
|
if len(out) != 0 {
|
|
t.Fatalf("acked → 0 repeats, got %+v", out)
|
|
}
|
|
if len(telegram.sends) != 0 {
|
|
t.Fatalf("acked → no telegram send, got %d", len(telegram.sends))
|
|
}
|
|
}
|
|
|
|
func TestRepeatUnackedSkipsBeforeInterval(t *testing.T) {
|
|
telegram := &fakeSink{}
|
|
ack := newFakeAck()
|
|
d := NewDispatcher(Config{Telegram: telegram, Ack: ack})
|
|
|
|
// sent 1min ago, interval 5min → wait.
|
|
_ = ack.MarkSent(context.Background(), "service_down", refNow().Add(-1*time.Minute))
|
|
|
|
out, err := d.RepeatUnacked(context.Background(), []string{"service_down"}, refNow(), 5*time.Minute, "down", "down")
|
|
if err != nil {
|
|
t.Fatalf("repeat: %v", err)
|
|
}
|
|
if len(out) != 0 {
|
|
t.Fatalf("before interval → 0 repeats, got %+v", out)
|
|
}
|
|
}
|
|
|
|
func TestRepeatUnackedNilTelegramOrAckIsNoOp(t *testing.T) {
|
|
d := NewDispatcher(Config{}) // no telegram, no ack
|
|
out, err := d.RepeatUnacked(context.Background(), []string{"service_down"}, refNow(), 5*time.Minute, "down", "down")
|
|
if err != nil {
|
|
t.Fatalf("nil telegram/ack: want nil err, got %v", err)
|
|
}
|
|
if out != nil {
|
|
t.Fatalf("nil telegram/ack: want nil, got %+v", out)
|
|
}
|
|
}
|
|
|
|
func TestDispatchRecurringReminderReschedules(t *testing.T) {
|
|
voice := &fakeSink{}
|
|
rc := &fakeReminderCompleter{}
|
|
d := NewDispatcher(Config{Voice: voice, Reminders: rc})
|
|
|
|
rd := loop.ReminderDecision{
|
|
Reminder: store.Reminder{ID: 7, Cron: "0 9 * * *", Status: "pending"},
|
|
State: loop.State{Now: refNow(), Presence: store.Present},
|
|
}
|
|
out, err := d.DispatchReminder(context.Background(), PhrasedReminder{
|
|
Decision: rd, Body: "daily standup", Summary: "standup",
|
|
}, refNow())
|
|
if err != nil {
|
|
t.Fatalf("dispatch: %v", err)
|
|
}
|
|
if len(out) != 1 {
|
|
t.Fatalf("want 1 dispatch, got %d", len(out))
|
|
}
|
|
// Should RescheduleReminder, not MarkReminder
|
|
if len(rc.rescheduled) != 1 || rc.rescheduled[0] != 7 {
|
|
t.Fatalf("want rescheduled 7, got %+v", rc.rescheduled)
|
|
}
|
|
if len(rc.marked) != 0 {
|
|
t.Fatalf("recurring reminder should not be marked fired: %+v", rc.marked)
|
|
}
|
|
}
|
|
|
|
// ----------------------------- durable outbox --------------------------------
|
|
|
|
type outboxAttempt struct {
|
|
kind, rule string
|
|
reminderID int64
|
|
channel, hash string
|
|
status string
|
|
begunAt, doneAt time.Time
|
|
}
|
|
|
|
// fakeOutbox — an in-memory Outbox that also lets a test simulate a crash
|
|
// exactly at the three points the audit finding cares about: before Begin
|
|
// records anything, after Begin but before the sink is called (the crash
|
|
// window that causes duplicate delivery — Begin already landed so recovery
|
|
// can see it), and after a successful send but before Complete persists the
|
|
// outcome (the other half of the same window).
|
|
type fakeOutbox struct {
|
|
attempts []*outboxAttempt
|
|
nextID int64
|
|
beginErr error
|
|
completeErr error
|
|
}
|
|
|
|
func (f *fakeOutbox) BeginDeliveryAttempt(_ context.Context, kind, rule string, reminderID int64, channel, hash string, now time.Time) (int64, error) {
|
|
if f.beginErr != nil {
|
|
return 0, f.beginErr
|
|
}
|
|
f.nextID++
|
|
f.attempts = append(f.attempts, &outboxAttempt{
|
|
kind: kind, rule: rule, reminderID: reminderID, channel: channel, hash: hash,
|
|
status: "pending", begunAt: now,
|
|
})
|
|
return f.nextID, nil
|
|
}
|
|
|
|
func (f *fakeOutbox) CompleteDeliveryAttempt(_ context.Context, id int64, status string, now time.Time) error {
|
|
if f.completeErr != nil {
|
|
return f.completeErr
|
|
}
|
|
if id < 1 || int(id) > len(f.attempts) {
|
|
return errors.New("fakeOutbox: unknown attempt id")
|
|
}
|
|
a := f.attempts[id-1]
|
|
a.status = status
|
|
a.doneAt = now
|
|
return nil
|
|
}
|
|
|
|
// TestDispatchNudge_OutboxRecordsBeforeSendAndCompletesAfter — the audit's
|
|
// core scenario: Begin must land BEFORE the sink is invoked (so a crash right
|
|
// after the external send but before Maven records anything still leaves a
|
|
// "pending" trail an operator/reconciler can find), and Complete must record
|
|
// the real outcome once Send returns.
|
|
func TestDispatchNudge_OutboxRecordsBeforeSendAndCompletesAfter(t *testing.T) {
|
|
voice := &fakeSink{}
|
|
rec := &fakeNudgeRecorder{}
|
|
ob := &fakeOutbox{}
|
|
d := NewDispatcher(Config{Voice: voice, Nudges: rec, Outbox: ob})
|
|
|
|
_, err := d.DispatchNudge(context.Background(), PhrasedNudge{
|
|
Candidate: candidate("water", loop.Sev1, store.Present),
|
|
Body: "you haven't had water in 4h",
|
|
Summary: "water",
|
|
}, refNow())
|
|
if err != nil {
|
|
t.Fatalf("dispatch: %v", err)
|
|
}
|
|
if len(ob.attempts) != 1 {
|
|
t.Fatalf("want 1 outbox attempt, got %d", len(ob.attempts))
|
|
}
|
|
a := ob.attempts[0]
|
|
if a.kind != "nudge" || a.rule != "water" || a.channel != "voice" {
|
|
t.Fatalf("unexpected attempt: %+v", a)
|
|
}
|
|
if a.status != store.DeliverySent {
|
|
t.Fatalf("want status sent after a successful send, got %q", a.status)
|
|
}
|
|
}
|
|
|
|
// TestDispatchNudge_OutboxMarksFailedOnSendError — a clean (non-crash) send
|
|
// error still gets a definite Complete("failed") — this is NOT the ambiguous
|
|
// case, so it must not be left "pending" for a reconciler to puzzle over.
|
|
func TestDispatchNudge_OutboxMarksFailedOnSendError(t *testing.T) {
|
|
voice := &fakeSink{err: errors.New("boom")}
|
|
ob := &fakeOutbox{}
|
|
d := NewDispatcher(Config{Voice: voice, Outbox: ob})
|
|
|
|
_, err := d.DispatchNudge(context.Background(), PhrasedNudge{
|
|
Candidate: candidate("water", loop.Sev1, store.Present),
|
|
Body: "body", Summary: "sum",
|
|
}, refNow())
|
|
if err == nil {
|
|
t.Fatal("expected send error to propagate")
|
|
}
|
|
if len(ob.attempts) != 1 || ob.attempts[0].status != store.DeliveryFailed {
|
|
t.Fatalf("want 1 attempt with status failed, got %+v", ob.attempts)
|
|
}
|
|
}
|
|
|
|
// TestDispatchReminder_OutboxTracksReminderAttempt — same Begin-before-Send /
|
|
// Complete-after-Send contract for the reminder path, which is a materially
|
|
// different code path (no Nudges recorder, MarkReminder/RescheduleReminder
|
|
// instead) and must not be skipped by only testing nudges.
|
|
func TestDispatchReminder_OutboxTracksReminderAttempt(t *testing.T) {
|
|
voice := &fakeSink{}
|
|
rc := &fakeReminderCompleter{}
|
|
ob := &fakeOutbox{}
|
|
d := NewDispatcher(Config{Voice: voice, Reminders: rc, Outbox: ob})
|
|
|
|
rd := loop.ReminderDecision{
|
|
Reminder: store.Reminder{ID: 42, Status: "pending"},
|
|
State: loop.State{Now: refNow(), Presence: store.Present},
|
|
}
|
|
_, err := d.DispatchReminder(context.Background(), PhrasedReminder{
|
|
Decision: rd, Body: "take out the trash", Summary: "trash",
|
|
}, refNow())
|
|
if err != nil {
|
|
t.Fatalf("dispatch: %v", err)
|
|
}
|
|
if len(ob.attempts) != 1 {
|
|
t.Fatalf("want 1 outbox attempt, got %d", len(ob.attempts))
|
|
}
|
|
a := ob.attempts[0]
|
|
if a.kind != "reminder" || a.reminderID != 42 || a.status != store.DeliverySent {
|
|
t.Fatalf("unexpected attempt: %+v", a)
|
|
}
|
|
}
|
|
|
|
// TestDispatchNudge_OutboxBeginFailureDoesNotBlockSend — losing outbox
|
|
// visibility on one attempt (e.g. a transient store error recording intent)
|
|
// must not itself prevent the nudge from reaching the user; the send still
|
|
// proceeds, just untracked for crash recovery on this one attempt.
|
|
func TestDispatchNudge_OutboxBeginFailureDoesNotBlockSend(t *testing.T) {
|
|
voice := &fakeSink{}
|
|
ob := &fakeOutbox{beginErr: errors.New("db busy")}
|
|
d := NewDispatcher(Config{Voice: voice, Outbox: ob})
|
|
|
|
out, err := d.DispatchNudge(context.Background(), PhrasedNudge{
|
|
Candidate: candidate("water", loop.Sev1, store.Present),
|
|
Body: "body", Summary: "sum",
|
|
}, refNow())
|
|
if err != nil {
|
|
t.Fatalf("dispatch: %v", err)
|
|
}
|
|
if len(voice.sends) != 1 || len(out) != 1 {
|
|
t.Fatalf("send should still happen despite outbox begin failure: sends=%v out=%v", voice.sends, out)
|
|
}
|
|
}
|