From b0f5a16ec9bff5b6c849f939acd1cecc488e0275 Mon Sep 17 00:00:00 2001 From: kami Date: Fri, 31 Jul 2026 23:09:22 +0400 Subject: [PATCH] Add digest as a real outcome: suppressed care nudges get resurfaced, not lost MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- cmd/mavend/digest_test.go | 158 ++++++++++++++++++++++++++ cmd/mavend/tick.go | 138 +++++++++++++++++++++++ internal/loop/gate_test.go | 48 ++++++++ internal/loop/loop.go | 30 +++++ internal/store/digest.go | 142 ++++++++++++++++++++++++ internal/store/digest_test.go | 202 ++++++++++++++++++++++++++++++++++ internal/store/migrations.go | 19 ++++ 7 files changed, 737 insertions(+) create mode 100644 cmd/mavend/digest_test.go create mode 100644 internal/store/digest.go create mode 100644 internal/store/digest_test.go diff --git a/cmd/mavend/digest_test.go b/cmd/mavend/digest_test.go new file mode 100644 index 0000000..3e808a9 --- /dev/null +++ b/cmd/mavend/digest_test.go @@ -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) + } +} diff --git a/cmd/mavend/tick.go b/cmd/mavend/tick.go index 39d1c83..495e2de 100644 --- a/cmd/mavend/tick.go +++ b/cmd/mavend/tick.go @@ -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. diff --git a/internal/loop/gate_test.go b/internal/loop/gate_test.go index ed0cb1c..8c66f45 100644 --- a/internal/loop/gate_test.go +++ b/internal/loop/gate_test.go @@ -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 diff --git a/internal/loop/loop.go b/internal/loop/loop.go index ae572d9..2070908 100644 --- a/internal/loop/loop.go +++ b/internal/loop/loop.go @@ -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. diff --git a/internal/store/digest.go b/internal/store/digest.go new file mode 100644 index 0000000..56838c9 --- /dev/null +++ b/internal/store/digest.go @@ -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 +} diff --git a/internal/store/digest_test.go b/internal/store/digest_test.go new file mode 100644 index 0000000..4d55af8 --- /dev/null +++ b/internal/store/digest_test.go @@ -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) + } + } +} diff --git a/internal/store/migrations.go b/internal/store/migrations.go index 8bc86eb..d59b54a 100644 --- a/internal/store/migrations.go +++ b/internal/store/migrations.go @@ -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