Add digest as a real outcome: suppressed care nudges get resurfaced, not lost
Vikunja #281. The interruption policy promised four outcomes — deliver_now, queue, digest, drop — but only three existed: a care candidate the restraint gate suppressed for quiet hours / away / calendar-busy simply vanished in loop.Tick's `continue`, with only the trace remembering why. internal/morning turned out not to be the natural drain: it's a fixed Item/FactKey checklist engine, not a generic message bundler, so gate- suppressed nudge text has nowhere to plug into its evidence model. Built a parallel (but small, reusing the outbox's shape) durable digest instead: - internal/store: digest_entries table + EnqueueDigestEntry (dedupes by rule+body, mirroring the delivery outbox's bodyHash), PendingDigestEntries, ExpireStaleDigestEntries, DrainDigestEntries (mark, never delete — an audit trail of what she actually said). - internal/loop: DigestEligible(severity, blockedBy) is the pure boundary — only genuine restraint blocks (quiet_hours/calendar_busy/presence) even qualify (cooldown/snooze are not "suppression"); within care, Sev2 (break) digests, Sev1 (water/meal — stale by the time anyone could resurface them) drops. High severity never digests; alarms bypass the gate and deliver unchanged, on purpose. - cmd/mavend/tick.go: each tick scans ExplainTick's trace for eligible blocked candidates, enqueues them, sweeps stale entries (24h expiry — the care rules are daily-cadence, so anything older is describing a day that's over), and drains the bundle only once the suppression reason has actually cleared, capped at 3 spoken items plus a trailing count so a digest can't turn into the exact nagging it was built to avoid. Tests: store-level round-trip/restart-survival/dedupe/expiry/drain, loop- level severity-boundary unit tests, and tick-level integration tests for the drain-only-when-clear and never-digest-high-severity behavior.
This commit is contained in:
kami
@@ -0,0 +1,158 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/kami/maven/internal/loop"
|
||||
"github.com/kami/maven/internal/store"
|
||||
)
|
||||
|
||||
// Vikunja #281 — the fourth delivery outcome: a care candidate the restraint
|
||||
// gate suppresses (quiet hours / away / calendar-busy) is not necessarily
|
||||
// lost. If it's worth resurfacing (loop.DigestEligible), it's durably held
|
||||
// (internal/store's digest_entries) and spoken as one bundle once speaking
|
||||
// is appropriate again — never while the suppression reason still holds.
|
||||
|
||||
func breakTrace(blockedBy string) *loop.TickTrace {
|
||||
return &loop.TickTrace{
|
||||
RuleTraces: []loop.RuleTrace{{
|
||||
RuleName: "break",
|
||||
Severity: loop.Sev2,
|
||||
PredicateResult: true,
|
||||
GateResult: false,
|
||||
GateBlockedBy: blockedBy,
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
// TestSuppressedCareDigestsAcrossQuietHours — a Sev2 care candidate blocked
|
||||
// by quiet hours is enqueued into the durable digest, and is spoken as a
|
||||
// "digest" nudge only once quiet hours actually end — never while still
|
||||
// suppressed (that would just be a second way to nag through quiet hours).
|
||||
func TestSuppressedCareDigestsAcrossQuietHours(t *testing.T) {
|
||||
st := newTestStore(t)
|
||||
sink := &fakeSink{}
|
||||
tl := newTestTickLoop(t, st, sink, nil)
|
||||
ctx := context.Background()
|
||||
now := refNow()
|
||||
|
||||
quiet := loop.State{Now: now, QuietHours: true, Presence: store.Present}
|
||||
tl.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, now)
|
||||
|
||||
entries, err := st.PendingDigestEntries(ctx, now)
|
||||
if err != nil {
|
||||
t.Fatalf("pending: %v", err)
|
||||
}
|
||||
if len(entries) != 1 || entries[0].Rule != "break" {
|
||||
t.Fatalf("want 1 pending digest entry for break, got %+v", entries)
|
||||
}
|
||||
|
||||
// still quiet hours: draining now must not speak — the same restraint
|
||||
// that suppressed the live nudge must suppress the bundle too.
|
||||
tl.maybeDrainDigest(ctx, quiet, now)
|
||||
if len(sink.sends) != 0 {
|
||||
t.Fatalf("digest must not drain while quiet hours holds, got %+v", sink.sends)
|
||||
}
|
||||
|
||||
// quiet hours end: this is the moment speaking is appropriate again.
|
||||
after := now.Add(time.Hour)
|
||||
clear := loop.State{Now: after, QuietHours: false, Presence: store.Present}
|
||||
tl.maybeDrainDigest(ctx, clear, after)
|
||||
|
||||
if len(sink.sends) != 1 {
|
||||
t.Fatalf("want exactly 1 dispatched digest bundle, got %d: %+v", len(sink.sends), sink.sends)
|
||||
}
|
||||
if sink.sends[0].RuleName != "digest" {
|
||||
t.Fatalf("want RuleName digest, got %q", sink.sends[0].RuleName)
|
||||
}
|
||||
|
||||
remaining, err := st.PendingDigestEntries(ctx, after)
|
||||
if err != nil {
|
||||
t.Fatalf("pending after drain: %v", err)
|
||||
}
|
||||
if len(remaining) != 0 {
|
||||
t.Fatalf("drained entry must no longer be pending, got %+v", remaining)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSuppressedCareDigestDedupesAcrossTicks — quiet hours holding for
|
||||
// several ticks must not enqueue several copies of the same suppressed
|
||||
// nudge; he hears it once when the bundle finally drains.
|
||||
func TestSuppressedCareDigestDedupesAcrossTicks(t *testing.T) {
|
||||
st := newTestStore(t)
|
||||
sink := &fakeSink{}
|
||||
tl := newTestTickLoop(t, st, sink, nil)
|
||||
ctx := context.Background()
|
||||
now := refNow()
|
||||
|
||||
quiet := loop.State{Now: now, QuietHours: true, Presence: store.Present}
|
||||
for i := 0; i < 3; i++ {
|
||||
tl.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, now.Add(time.Duration(i)*time.Minute))
|
||||
}
|
||||
|
||||
entries, err := st.PendingDigestEntries(ctx, now)
|
||||
if err != nil {
|
||||
t.Fatalf("pending: %v", err)
|
||||
}
|
||||
if len(entries) != 1 {
|
||||
t.Fatalf("3 suppressions of the same nudge must collapse to 1 pending entry, got %d", len(entries))
|
||||
}
|
||||
}
|
||||
|
||||
// TestSuppressedCareDigestExpiresRatherThanDeliveringLate — an entry that
|
||||
// aged out before the suppression cleared is dropped, not spoken late: a
|
||||
// two-day-old "you skipped a break" is noise, not news.
|
||||
func TestSuppressedCareDigestExpiresRatherThanDeliveringLate(t *testing.T) {
|
||||
st := newTestStore(t)
|
||||
sink := &fakeSink{}
|
||||
tl := newTestTickLoop(t, st, sink, nil)
|
||||
ctx := context.Background()
|
||||
now := refNow()
|
||||
|
||||
quiet := loop.State{Now: now, QuietHours: true, Presence: store.Present}
|
||||
tl.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, now)
|
||||
|
||||
// well past digestExpiry (24h) before the suppression ever clears.
|
||||
stale := now.Add(48 * time.Hour)
|
||||
tl.expireStaleDigest(ctx, stale)
|
||||
|
||||
clear := loop.State{Now: stale, QuietHours: false, Presence: store.Present}
|
||||
tl.maybeDrainDigest(ctx, clear, stale)
|
||||
|
||||
if len(sink.sends) != 0 {
|
||||
t.Fatalf("a stale digest entry must be dropped, not delivered late; got %+v", sink.sends)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSuppressedCareDigestIgnoresHighSeverity — defense in depth at the
|
||||
// wiring layer: even if a RuleTrace somehow showed a high-severity rule
|
||||
// blocked by a care-only gate reason, the tick driver must not durably
|
||||
// digest it. Alarms bypass the gate and deliver now, unchanged; they must
|
||||
// never be silently delayed into a bundle.
|
||||
func TestSuppressedCareDigestIgnoresHighSeverity(t *testing.T) {
|
||||
st := newTestStore(t)
|
||||
sink := &fakeSink{}
|
||||
tl := newTestTickLoop(t, st, sink, nil)
|
||||
ctx := context.Background()
|
||||
now := refNow()
|
||||
|
||||
trace := &loop.TickTrace{RuleTraces: []loop.RuleTrace{{
|
||||
RuleName: "service_down",
|
||||
Severity: loop.Sev4,
|
||||
PredicateResult: true,
|
||||
GateResult: false,
|
||||
GateBlockedBy: "quiet_hours",
|
||||
}}}
|
||||
quiet := loop.State{Now: now, QuietHours: true, Presence: store.Present}
|
||||
tl.enqueueSuppressedDigest(ctx, trace, quiet, now)
|
||||
|
||||
entries, err := st.PendingDigestEntries(ctx, now)
|
||||
if err != nil {
|
||||
t.Fatalf("pending: %v", err)
|
||||
}
|
||||
if len(entries) != 0 {
|
||||
t.Fatalf("high severity must never be digested, got %+v", entries)
|
||||
}
|
||||
}
|
||||
@@ -180,6 +180,17 @@ func (t *tickLoop) tick(ctx context.Context, now time.Time) {
|
||||
// (with dedup) avoids re-queueing the same rule after a flush.
|
||||
t.maybeFlush(ctx, now, state)
|
||||
|
||||
// gate-suppressed digest (Vikunja #281): rules the restraint gate held
|
||||
// back this tick (quiet hours / away / calendar-busy), not because they
|
||||
// weren't due, but because it wasn't the moment. Some of those are worth
|
||||
// resurfacing later instead of just being lost — loop.DigestEligible
|
||||
// draws that line. This is a SEPARATE mechanism from the in-memory
|
||||
// digestQ above: that one batches candidates the gate already ALLOWED to
|
||||
// fire; this one durably holds candidates the gate BLOCKED.
|
||||
t.enqueueSuppressedDigest(ctx, trace, state, now)
|
||||
t.expireStaleDigest(ctx, now)
|
||||
t.maybeDrainDigest(ctx, state, now)
|
||||
|
||||
// routines: operator-declared scheduled behaviors. fire the ones whose cron
|
||||
// crossed since last fire, delivered through the normal routing (voice when
|
||||
// present, away channels otherwise). bodies are literal operator text — not
|
||||
@@ -342,6 +353,133 @@ func (t *tickLoop) flushDigest(ctx context.Context, now time.Time, state loop.St
|
||||
t.digestQ = nil
|
||||
}
|
||||
|
||||
// digestExpiry — how long a gate-suppressed care nudge stays worth
|
||||
// resurfacing. 24h: these are daily-cadence rules (water/meal/break run on
|
||||
// hour-scale cooldowns and re-derive from facts that reset every day), so a
|
||||
// digest entry that outlives one full day is describing a day that's already
|
||||
// over — "you skipped a break yesterday" said tomorrow evening is noise, not
|
||||
// news. Bounding at one day also means a digest can never silently span a
|
||||
// weekend of quiet hours into an unbounded backlog.
|
||||
const digestExpiry = 24 * time.Hour
|
||||
|
||||
// maxDigestSpokenItems — the bundle read-out is capped so "batched, not
|
||||
// dropped" cannot regress into "she dumps twelve things on me the moment I
|
||||
// walk in" — a digest that nags in bulk is worse than the drops it replaced.
|
||||
// Anything beyond the cap is still marked drained (it did get its moment;
|
||||
// the cap limits WORDS, not whether it counted) and folded into a trailing
|
||||
// count instead of being spoken in full.
|
||||
const maxDigestSpokenItems = 3
|
||||
|
||||
// enqueueSuppressedDigest scans this tick's trace for care candidates the
|
||||
// gate blocked for a genuine restraint reason and durably records the
|
||||
// digest-eligible ones (loop.DigestEligible). Phrasing happens once, here,
|
||||
// at enqueue time — not re-derived at drain time — the same way queueNudge
|
||||
// phrases once and caches, so a rule suppressed for hours isn't re-prompting
|
||||
// the LLM every tick it stays blocked (EnqueueDigestEntry's rule+body dedupe
|
||||
// makes repeat calls here harmless, but skipping the phrase call entirely
|
||||
// when a pending entry already exists avoids the LLM round-trip too).
|
||||
func (t *tickLoop) enqueueSuppressedDigest(ctx context.Context, trace *loop.TickTrace, state loop.State, now time.Time) {
|
||||
if trace == nil {
|
||||
return
|
||||
}
|
||||
for _, tr := range trace.RuleTraces {
|
||||
if !tr.PredicateResult || tr.GateResult {
|
||||
continue // didn't want to fire, or wasn't suppressed
|
||||
}
|
||||
if !loop.DigestEligible(tr.Severity, tr.GateBlockedBy) {
|
||||
continue
|
||||
}
|
||||
rule := loop.Rule{Name: tr.RuleName, Severity: tr.Severity}
|
||||
cand := loop.Candidate{Rule: rule, Severity: tr.Severity, State: state}
|
||||
pn, err := t.phraser.PhraseNudge(ctx, cand)
|
||||
if err != nil {
|
||||
log.Printf("tick: phrase digest candidate %s: %v", tr.RuleName, err)
|
||||
continue
|
||||
}
|
||||
expires := now.Add(digestExpiry)
|
||||
if _, deduped, err := t.store.EnqueueDigestEntry(ctx, tr.RuleName, int(tr.Severity), pn.Body, now, expires); err != nil {
|
||||
log.Printf("tick: enqueue digest entry %s: %v", tr.RuleName, err)
|
||||
} else if deduped {
|
||||
// same suppressed nudge already pending — nothing new to say.
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// expireStaleDigest sweeps entries past their expiry once per tick — cheap
|
||||
// bookkeeping, mirrors ReconcileStaleDeliveryAttempts's shape.
|
||||
func (t *tickLoop) expireStaleDigest(ctx context.Context, now time.Time) {
|
||||
n, err := t.store.ExpireStaleDigestEntries(ctx, now)
|
||||
if err != nil {
|
||||
log.Printf("tick: expire stale digest entries: %v", err)
|
||||
return
|
||||
}
|
||||
if n > 0 {
|
||||
log.Printf("tick: expired %d stale digest entr(y/ies) unspoken", n)
|
||||
}
|
||||
}
|
||||
|
||||
// maybeDrainDigest speaks the pending digest bundle once the gate's
|
||||
// suppression reasons have actually cleared — quiet hours over, back from
|
||||
// away, out of the meeting. Draining while still suppressed would just be a
|
||||
// second way to nag through quiet hours; the bundle waits for the same "is
|
||||
// it allowed right now" condition a live nudge already waits for.
|
||||
func (t *tickLoop) maybeDrainDigest(ctx context.Context, state loop.State, now time.Time) {
|
||||
if state.QuietHours || state.CalendarBusy || state.Presence == store.Away {
|
||||
return
|
||||
}
|
||||
entries, err := t.store.PendingDigestEntries(ctx, now)
|
||||
if err != nil {
|
||||
log.Printf("tick: pending digest entries: %v", err)
|
||||
return
|
||||
}
|
||||
if len(entries) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
spoken := entries
|
||||
extra := 0
|
||||
if len(spoken) > maxDigestSpokenItems {
|
||||
spoken = entries[:maxDigestSpokenItems]
|
||||
extra = len(entries) - maxDigestSpokenItems
|
||||
}
|
||||
var b strings.Builder
|
||||
maxSev := 0
|
||||
for i, e := range spoken {
|
||||
if i > 0 {
|
||||
b.WriteString(" · ")
|
||||
}
|
||||
b.WriteString(e.Body)
|
||||
if e.Severity > maxSev {
|
||||
maxSev = e.Severity
|
||||
}
|
||||
}
|
||||
if extra > 0 {
|
||||
fmt.Fprintf(&b, " · и ещё %d", extra)
|
||||
}
|
||||
body := b.String()
|
||||
summary := fmt.Sprintf("%d отложенных уведомлений", len(entries))
|
||||
|
||||
cand := loop.Candidate{
|
||||
Rule: loop.Rule{Name: "digest", Severity: loop.Severity(maxSev)},
|
||||
Severity: loop.Severity(maxSev),
|
||||
State: state,
|
||||
}
|
||||
pn := delivery.PhrasedNudge{Candidate: cand, Body: body, Summary: summary}
|
||||
t.cachePhrase(pn)
|
||||
if _, err := t.dispatcher.DispatchNudge(ctx, pn, now); err != nil {
|
||||
log.Printf("tick: dispatch digest bundle: %v", err)
|
||||
return // leave entries pending; retried next tick
|
||||
}
|
||||
ids := make([]int64, len(entries))
|
||||
for i, e := range entries {
|
||||
ids[i] = e.ID
|
||||
}
|
||||
if err := t.store.DrainDigestEntries(ctx, ids, now); err != nil {
|
||||
log.Printf("tick: drain digest entries: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// routinesFromConfig maps the config's routine blocks to the engine type.
|
||||
// Validation (cron parses, name/body present, severity defaulted) already ran
|
||||
// in config.Load, so this is a pure field copy.
|
||||
|
||||
@@ -286,6 +286,54 @@ func TestRemindersStillHonourSnooze(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------- digest eligibility ------------------------------
|
||||
|
||||
// A Sev2 care candidate (break) suppressed for a genuine restraint reason is
|
||||
// worth resurfacing later.
|
||||
func TestDigestEligibleSev2SuppressedByRestraint(t *testing.T) {
|
||||
for _, reason := range []string{"quiet_hours", "calendar_busy", "presence"} {
|
||||
if !DigestEligible(Sev2, reason) {
|
||||
t.Errorf("sev2 blocked by %q: want digest-eligible", reason)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A Sev1 care candidate (water/meal) never digests — a biological timer
|
||||
// nudge is stale by the time anyone could resurface it, so it just drops.
|
||||
func TestDigestEligibleSev1NeverDigests(t *testing.T) {
|
||||
for _, reason := range []string{"quiet_hours", "calendar_busy", "presence"} {
|
||||
if DigestEligible(Sev1, reason) {
|
||||
t.Errorf("sev1 blocked by %q: want drop, got digest-eligible", reason)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Ops severities are never blocked by these reasons in practice (Gate only
|
||||
// applies quiet_hours/calendar_busy/presence to care severities), but the
|
||||
// boundary itself must refuse to digest a high severity even if asked —
|
||||
// alarms bypass the gate and deliver now, unchanged, never delayed.
|
||||
func TestDigestEligibleNeverDigestsHighSeverity(t *testing.T) {
|
||||
for _, sev := range []Severity{Sev3, Sev4} {
|
||||
for _, reason := range []string{"quiet_hours", "calendar_busy", "presence"} {
|
||||
if DigestEligible(sev, reason) {
|
||||
t.Errorf("sev%d blocked by %q: high severity must never digest", sev, reason)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// cooldown and snooze are not "suppression" in the digest sense — cooldown
|
||||
// means it was already said recently, snooze means the user asked to not
|
||||
// hear about it. Neither should resurface later just because the severity
|
||||
// matches.
|
||||
func TestDigestEligibleExcludesCooldownAndSnooze(t *testing.T) {
|
||||
for _, reason := range []string{"cooldown", "snooze", "inert_no_data", "predicate", ""} {
|
||||
if DigestEligible(Sev2, reason) {
|
||||
t.Errorf("sev2 blocked by %q: should not be digest-eligible", reason)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// GAP — the gate reads State.SnoozeUntil, but the Gatherer hard-codes it to nil
|
||||
// (internal/loop/gather.go:153), so snooze is dead in the running daemon: the
|
||||
// unit tests above pass while nothing can ever populate the map. This asserts
|
||||
|
||||
@@ -105,6 +105,36 @@ func Tick(s State, rules []Rule) *Candidate {
|
||||
return fire
|
||||
}
|
||||
|
||||
// DigestEligible decides digest-vs-drop for a care candidate the gate
|
||||
// suppressed this tick (see ExplainGate's blockedBy). Pure — no I/O, no
|
||||
// state, just the two facts that matter: why it was suppressed, and how
|
||||
// insistent it was.
|
||||
//
|
||||
// Only genuine RESTRAINT blocks are eligible at all — quiet_hours,
|
||||
// calendar_busy, presence(away). cooldown and snooze are not suppression in
|
||||
// this sense: cooldown means "you already heard this recently" (resurfacing
|
||||
// it later would be an actual repeat, not a rescue) and snooze is the user
|
||||
// explicitly saying "not this" (digesting it anyway would defeat the ask).
|
||||
// Ops severities (Sev3/4) never reach here — the gate never blocks them for
|
||||
// these reasons in the first place (see Gate), and even if a future rule
|
||||
// dropped Sev3+ into "care", digest still refuses them: alarms bypass the
|
||||
// gate on purpose and must never be silently delayed into a bundle.
|
||||
//
|
||||
// Within care (Sev1–2), the boundary is severity itself: Sev1 (water, meal —
|
||||
// biological timers with no "still relevant later" property; a water nudge
|
||||
// from 3 hours into quiet hours is just wrong by morning) drops. Sev2
|
||||
// (break — "you worked through a long stretch without a break while I
|
||||
// couldn't reach you") is information that stays true and useful after the
|
||||
// fact, so it digests.
|
||||
func DigestEligible(sev Severity, blockedBy string) bool {
|
||||
switch blockedBy {
|
||||
case "quiet_hours", "calendar_busy", "presence":
|
||||
default:
|
||||
return false
|
||||
}
|
||||
return sev == Sev2
|
||||
}
|
||||
|
||||
// ReminderDecision — a due reminder the daemon should deliver now.
|
||||
// NOT gated by the universal Gate (per spec: "wake me 7" fires in quiet hours;
|
||||
// that's the point). Snooze is the one part of restraint that still applies.
|
||||
|
||||
@@ -0,0 +1,142 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Digest entry statuses. pending = enqueued, waiting for a drain. drained =
|
||||
// spoken as part of a bundle. expired = the tick loop's expiry sweep found it
|
||||
// past its expires_ts before a drain happened — dropped, not delivered late.
|
||||
const (
|
||||
DigestPending = "pending"
|
||||
DigestDrained = "drained"
|
||||
DigestExpired = "expired"
|
||||
)
|
||||
|
||||
// DigestEntry — one gate-suppressed care candidate durably held for later
|
||||
// bundled delivery.
|
||||
type DigestEntry struct {
|
||||
ID int64
|
||||
Rule string
|
||||
Severity int
|
||||
Body string
|
||||
CreatedTs time.Time
|
||||
ExpiresTs time.Time
|
||||
}
|
||||
|
||||
// DigestBodyHash is the dedupe key for a digest entry: same rule, same
|
||||
// wording ⇒ the same suppressed nudge repeating across ticks, and he should
|
||||
// hear it once, not once per tick it kept getting suppressed.
|
||||
func DigestBodyHash(rule, body string) string {
|
||||
sum := sha256.Sum256([]byte(rule + "\x00" + body))
|
||||
return hex.EncodeToString(sum[:8])
|
||||
}
|
||||
|
||||
// EnqueueDigestEntry durably records a suppressed care candidate worth
|
||||
// resurfacing later. If a pending entry with the same rule+body already
|
||||
// exists, this is a no-op that returns the existing id and deduped=true —
|
||||
// the same suppressed nudge repeating across ticks must not pile up into
|
||||
// several copies of itself in the eventual bundle.
|
||||
func (s *Store) EnqueueDigestEntry(ctx context.Context, rule string, severity int, body string, now, expiresAt time.Time) (id int64, deduped bool, err error) {
|
||||
hash := DigestBodyHash(rule, body)
|
||||
var existing int64
|
||||
err = s.db.QueryRowContext(ctx,
|
||||
`SELECT id FROM digest_entries WHERE status = ? AND rule = ? AND body_hash = ? LIMIT 1`,
|
||||
DigestPending, rule, hash).Scan(&existing)
|
||||
if err == nil {
|
||||
return existing, true, nil
|
||||
}
|
||||
|
||||
res, err := s.db.ExecContext(ctx,
|
||||
`INSERT INTO digest_entries (rule, severity, body, body_hash, status, created_ts, expires_ts)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)`,
|
||||
rule, severity, body, hash, DigestPending, now.UnixMilli(), expiresAt.UnixMilli())
|
||||
if err != nil {
|
||||
return 0, false, fmt.Errorf("enqueue digest entry: %w", err)
|
||||
}
|
||||
id, err = res.LastInsertId()
|
||||
if err != nil {
|
||||
return 0, false, fmt.Errorf("enqueue digest entry: last insert id: %w", err)
|
||||
}
|
||||
return id, false, nil
|
||||
}
|
||||
|
||||
// PendingDigestEntries returns the live (not yet expired) pending entries,
|
||||
// oldest first — the order they were suppressed in, which is also the order
|
||||
// a bundled readout should mention them.
|
||||
func (s *Store) PendingDigestEntries(ctx context.Context, now time.Time) ([]DigestEntry, error) {
|
||||
rows, err := s.db.QueryContext(ctx,
|
||||
`SELECT id, rule, severity, body, created_ts, expires_ts
|
||||
FROM digest_entries WHERE status = ? AND expires_ts > ? ORDER BY created_ts ASC`,
|
||||
DigestPending, now.UnixMilli())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("pending digest entries: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []DigestEntry
|
||||
for rows.Next() {
|
||||
var e DigestEntry
|
||||
var created, expires int64
|
||||
if err := rows.Scan(&e.ID, &e.Rule, &e.Severity, &e.Body, &created, &expires); err != nil {
|
||||
return nil, fmt.Errorf("pending digest entries: scan: %w", err)
|
||||
}
|
||||
e.CreatedTs = time.UnixMilli(created)
|
||||
e.ExpiresTs = time.UnixMilli(expires)
|
||||
out = append(out, e)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// ExpireStaleDigestEntries marks pending entries whose expires_ts has passed
|
||||
// as expired — stale information (yesterday's battery warning) is noise, not
|
||||
// news, so it is dropped rather than delivered late. Called once per tick,
|
||||
// mirroring ReconcileStaleDeliveryAttempts's "sweep, don't guess" shape.
|
||||
// Returns the count expired, for logging.
|
||||
func (s *Store) ExpireStaleDigestEntries(ctx context.Context, now time.Time) (int, error) {
|
||||
res, err := s.db.ExecContext(ctx,
|
||||
`UPDATE digest_entries SET status = ? WHERE status = ? AND expires_ts <= ?`,
|
||||
DigestExpired, DigestPending, now.UnixMilli())
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("expire stale digest entries: %w", err)
|
||||
}
|
||||
n, err := res.RowsAffected()
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("expire stale digest entries: rows affected: %w", err)
|
||||
}
|
||||
return int(n), nil
|
||||
}
|
||||
|
||||
// DrainDigestEntries marks the given entries drained — they were folded into
|
||||
// a bundle that was successfully dispatched. Called only after a successful
|
||||
// send, same rule as the delivery outbox: a failed dispatch must not mark
|
||||
// entries drained, or the bundle is lost along with the failed send.
|
||||
func (s *Store) DrainDigestEntries(ctx context.Context, ids []int64, now time.Time) error {
|
||||
if len(ids) == 0 {
|
||||
return nil
|
||||
}
|
||||
tx, err := s.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("drain digest entries: begin: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
stmt, err := tx.PrepareContext(ctx,
|
||||
`UPDATE digest_entries SET status = ? WHERE id = ? AND status = ?`)
|
||||
if err != nil {
|
||||
return fmt.Errorf("drain digest entries: prepare: %w", err)
|
||||
}
|
||||
defer stmt.Close()
|
||||
for _, id := range ids {
|
||||
if _, err := stmt.ExecContext(ctx, DigestDrained, id, DigestPending); err != nil {
|
||||
return fmt.Errorf("drain digest entry %d: %w", id, err)
|
||||
}
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return fmt.Errorf("drain digest entries: commit: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,202 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// TestDigestEntryRoundTrips — a suppressed care candidate lands durably and
|
||||
// comes back out of PendingDigestEntries with its severity and body intact.
|
||||
func TestDigestEntryRoundTrips(t *testing.T) {
|
||||
s := newTestStore(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now()
|
||||
|
||||
id, deduped, err := s.EnqueueDigestEntry(ctx, "break", 2, "ты долго не отдыхала", now, now.Add(24*time.Hour))
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue: %v", err)
|
||||
}
|
||||
if deduped {
|
||||
t.Fatal("first enqueue must not report deduped")
|
||||
}
|
||||
if id == 0 {
|
||||
t.Fatal("want a nonzero id")
|
||||
}
|
||||
|
||||
entries, err := s.PendingDigestEntries(ctx, now)
|
||||
if err != nil {
|
||||
t.Fatalf("pending: %v", err)
|
||||
}
|
||||
if len(entries) != 1 || entries[0].ID != id {
|
||||
t.Fatalf("want 1 pending entry with id %d, got %+v", id, entries)
|
||||
}
|
||||
if entries[0].Rule != "break" || entries[0].Severity != 2 || entries[0].Body != "ты долго не отдыхала" {
|
||||
t.Fatalf("entry contents wrong: %+v", entries[0])
|
||||
}
|
||||
}
|
||||
|
||||
// TestDigestEntrySurvivesRestart — durability is the whole point: a fresh
|
||||
// Store handle on the same file must see the same pending entry, exactly
|
||||
// like the delivery outbox's crash-recovery promise.
|
||||
func TestDigestEntrySurvivesRestart(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
ctx := context.Background()
|
||||
now := time.Now()
|
||||
|
||||
s1, err := Open(ctx, dir+"/m.db")
|
||||
if err != nil {
|
||||
t.Fatalf("open: %v", err)
|
||||
}
|
||||
id, _, err := s1.EnqueueDigestEntry(ctx, "break", 2, "перерыв", now, now.Add(24*time.Hour))
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue: %v", err)
|
||||
}
|
||||
if err := s1.Close(); err != nil {
|
||||
t.Fatalf("close: %v", err)
|
||||
}
|
||||
|
||||
// simulated restart: a brand new Store handle on the same file.
|
||||
s2, err := Open(ctx, dir+"/m.db")
|
||||
if err != nil {
|
||||
t.Fatalf("reopen: %v", err)
|
||||
}
|
||||
defer func() { _ = s2.Close() }()
|
||||
|
||||
entries, err := s2.PendingDigestEntries(ctx, now)
|
||||
if err != nil {
|
||||
t.Fatalf("pending after restart: %v", err)
|
||||
}
|
||||
if len(entries) != 1 || entries[0].ID != id {
|
||||
t.Fatalf("digest entry did not survive restart: %+v", entries)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDigestEntryDedupesSameRuleAndBody — the same suppressed nudge
|
||||
// repeating across ticks (quiet hours holding for hours) must not pile up
|
||||
// into several copies of itself; he hears it once.
|
||||
func TestDigestEntryDedupesSameRuleAndBody(t *testing.T) {
|
||||
s := newTestStore(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now()
|
||||
|
||||
id1, deduped1, err := s.EnqueueDigestEntry(ctx, "break", 2, "перерыв нужен", now, now.Add(24*time.Hour))
|
||||
if err != nil {
|
||||
t.Fatalf("first enqueue: %v", err)
|
||||
}
|
||||
if deduped1 {
|
||||
t.Fatal("first enqueue should not be deduped")
|
||||
}
|
||||
|
||||
for i := 0; i < 2; i++ {
|
||||
id2, deduped2, err := s.EnqueueDigestEntry(ctx, "break", 2, "перерыв нужен", now.Add(time.Minute), now.Add(25*time.Hour))
|
||||
if err != nil {
|
||||
t.Fatalf("repeat enqueue: %v", err)
|
||||
}
|
||||
if !deduped2 {
|
||||
t.Fatal("repeat enqueue of the same rule+body should report deduped")
|
||||
}
|
||||
if id2 != id1 {
|
||||
t.Fatalf("deduped enqueue should return the original id: want %d got %d", id1, id2)
|
||||
}
|
||||
}
|
||||
|
||||
entries, err := s.PendingDigestEntries(ctx, now)
|
||||
if err != nil {
|
||||
t.Fatalf("pending: %v", err)
|
||||
}
|
||||
if len(entries) != 1 {
|
||||
t.Fatalf("want exactly 1 pending entry after 3 enqueues of the same nudge, got %d", len(entries))
|
||||
}
|
||||
}
|
||||
|
||||
// TestDigestEntryExpiresRatherThanDeliversLate — a stale entry (past its
|
||||
// expires_ts) must not surface in PendingDigestEntries, and the sweep should
|
||||
// mark it expired instead of leaving it around to be delivered late.
|
||||
func TestDigestEntryExpiresRatherThanDeliversLate(t *testing.T) {
|
||||
s := newTestStore(t)
|
||||
ctx := context.Background()
|
||||
created := time.Now()
|
||||
expiresAt := created.Add(time.Hour)
|
||||
|
||||
id, _, err := s.EnqueueDigestEntry(ctx, "water", 1, "стакан воды", created, expiresAt)
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue: %v", err)
|
||||
}
|
||||
|
||||
afterExpiry := expiresAt.Add(time.Minute)
|
||||
|
||||
// even before the sweep runs, a stale entry must not be handed back as
|
||||
// pending — "not yet swept" must not mean "still deliverable".
|
||||
entries, err := s.PendingDigestEntries(ctx, afterExpiry)
|
||||
if err != nil {
|
||||
t.Fatalf("pending: %v", err)
|
||||
}
|
||||
if len(entries) != 0 {
|
||||
t.Fatalf("stale entry must not be returned as pending, got %+v", entries)
|
||||
}
|
||||
|
||||
n, err := s.ExpireStaleDigestEntries(ctx, afterExpiry)
|
||||
if err != nil {
|
||||
t.Fatalf("expire sweep: %v", err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("want 1 entry expired, got %d", n)
|
||||
}
|
||||
|
||||
var status string
|
||||
if err := s.db.QueryRowContext(ctx, `SELECT status FROM digest_entries WHERE id = ?`, id).Scan(&status); err != nil {
|
||||
t.Fatalf("read back: %v", err)
|
||||
}
|
||||
if status != DigestExpired {
|
||||
t.Fatalf("status: want %q, got %q", DigestExpired, status)
|
||||
}
|
||||
|
||||
// idempotent: a second sweep finds nothing new.
|
||||
n2, err := s.ExpireStaleDigestEntries(ctx, afterExpiry.Add(time.Hour))
|
||||
if err != nil {
|
||||
t.Fatalf("second sweep: %v", err)
|
||||
}
|
||||
if n2 != 0 {
|
||||
t.Fatalf("second sweep should find nothing, got %d", n2)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDigestEntryDrainMarksDrainedNotDeleted — draining is bookkeeping, not
|
||||
// deletion: the row survives as an audit trail of what she actually said.
|
||||
func TestDigestEntryDrainMarksDrainedNotDeleted(t *testing.T) {
|
||||
s := newTestStore(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now()
|
||||
|
||||
id1, _, err := s.EnqueueDigestEntry(ctx, "break", 2, "перерыв", now, now.Add(24*time.Hour))
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue 1: %v", err)
|
||||
}
|
||||
id2, _, err := s.EnqueueDigestEntry(ctx, "break2", 2, "другое", now, now.Add(24*time.Hour))
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue 2: %v", err)
|
||||
}
|
||||
|
||||
if err := s.DrainDigestEntries(ctx, []int64{id1, id2}, now.Add(time.Hour)); err != nil {
|
||||
t.Fatalf("drain: %v", err)
|
||||
}
|
||||
|
||||
entries, err := s.PendingDigestEntries(ctx, now.Add(time.Hour))
|
||||
if err != nil {
|
||||
t.Fatalf("pending: %v", err)
|
||||
}
|
||||
if len(entries) != 0 {
|
||||
t.Fatalf("drained entries must not still be pending, got %+v", entries)
|
||||
}
|
||||
|
||||
for _, id := range []int64{id1, id2} {
|
||||
var status string
|
||||
if err := s.db.QueryRowContext(ctx, `SELECT status FROM digest_entries WHERE id = ?`, id).Scan(&status); err != nil {
|
||||
t.Fatalf("read back %d: %v", id, err)
|
||||
}
|
||||
if status != DigestDrained {
|
||||
t.Fatalf("entry %d status: want %q, got %q", id, DigestDrained, status)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -112,6 +112,25 @@ ALTER TABLE reminders ADD COLUMN next_fire_ts INTEGER;`, // #2
|
||||
DROP TABLE delivery_attempts;
|
||||
ALTER TABLE delivery_attempts_v12 RENAME TO delivery_attempts;
|
||||
CREATE INDEX IF NOT EXISTS idx_delivery_attempts_status ON delivery_attempts (status);`,
|
||||
|
||||
// #13 — durable digest outbox (Vikunja #281). A care nudge the restraint
|
||||
// gate suppresses (quiet hours / away / calendar-busy) is not necessarily
|
||||
// lost: if it's worth resurfacing, it lands here instead, and gets spoken
|
||||
// as one bundle at the next moment speaking is appropriate. body_hash
|
||||
// dedupes repeat suppressions of the "same" nudge; expires_ts bounds how
|
||||
// stale an entry may get before it's worthless and must be dropped rather
|
||||
// than delivered late.
|
||||
`CREATE TABLE IF NOT EXISTS digest_entries (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
rule TEXT NOT NULL,
|
||||
severity INTEGER NOT NULL,
|
||||
body TEXT NOT NULL,
|
||||
body_hash TEXT NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','drained','expired')),
|
||||
created_ts INTEGER NOT NULL,
|
||||
expires_ts INTEGER NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_digest_entries_status ON digest_entries (status);`,
|
||||
}
|
||||
|
||||
// migrate applies every migration with a number greater than the DB's current
|
||||
|
||||
Reference in New Issue
Block a user