Merge the five small-task branches

2026-07-31 23:11:26 +04:00
parent 7ad3c9a408 b0f5a16ec9
commit 9b8bdf73cc
7 changed files with 737 additions and 0 deletions
@@ -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