store: the first repeat-til-ack send, and a digest entry that expired unswept (V-617)

LastSent scanned MAX(sent_at) over an empty ack_sends into a bare int64, so
the ordinary "nothing sent yet" case came back as a scan error rather than the
zero time its doc promises. ack_sends is written only by MarkSent, and MarkSent
runs only after a repeat has gone out, so every rule's FIRST repeat read an
empty table — and RepeatUnacked aborts its whole sweep on that error. The
repeat-til-ack loop could never take its first step. Scans into a NullInt64,
the same way OldestPendingTelegram already does two files over.

EnqueueDigestEntry deduped against any row still marked pending, including one
already past its expires_ts. The tick enqueues before it sweeps, so a suppressed
nudge arriving on the tick after an expiry was told deduped=true against an
entry PendingDigestEntries will never hand back: the caller drops the phrasing
it just paid the LLM for and nothing reaches the bundle. The dedupe now carries
the same expiry test the read side does. Its lookup also stops treating a real
read failure as "nothing there".

internal/store/ack.go

@@ -2,6 +2,7 @@ package store
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
 	"time"
 )
@@ -38,16 +39,21 @@ func (s *Store) MarkSent(ctx context.Context, key string, ts time.Time) error {
 // send goes through RecordNudge, not MarkSent, so the first MarkSent comes on
 // the repeat path — LastSent may legitimately be zero until then).
 func (s *Store) LastSent(ctx context.Context, key string) (time.Time, error) {
-	var millis int64
+	// MAX over an empty set is one row holding NULL, not zero rows, so this
+	// scans into a NullInt64 — the same trap OldestPendingTelegram spells out.
+	// A bare int64 turned the ordinary "nothing sent yet" case into a scan
+	// error, and RepeatUnacked aborts its whole sweep on one, so the first
+	// repeat could never go out for any rule.
+	var millis sql.NullInt64
 	err := s.db.QueryRowContext(ctx,
 		`SELECT MAX(sent_at) FROM ack_sends WHERE rule = ?`, key).Scan(&millis)
 	if err != nil {
 		return time.Time{}, fmt.Errorf("last sent %s: %w", key, err)
 	}
-	if millis == 0 {
+	if !millis.Valid {
 		return time.Time{}, nil
 	}
-	return time.UnixMilli(millis).UTC(), nil
+	return time.UnixMilli(millis.Int64).UTC(), nil
 }
 
 // MarkAcked marks ALL pending telegram nudges for the rule as "acted" —

internal/store/ack_test.go

@@ -0,0 +1,60 @@
+package store
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// TestLastSentOnEmptyTableIsZero pins the aggregate-over-nothing trap that
+// nudges.go's OldestPendingTelegram already documents: MAX over an empty set is
+// one row holding NULL, not zero rows. Scanning that into a bare int64 is an
+// error, and the doc on LastSent promises a zero time instead.
+//
+// It is not a cosmetic promise. ack_sends is written only by MarkSent, and
+// MarkSent is called only after a repeat has already gone out, so the first
+// repeat for every rule reads an empty table. delivery.Dispatcher.RepeatUnacked
+// aborts the whole sweep on that error, which means the repeat-til-ack loop can
+// never take its first step for any rule.
+func TestLastSentOnEmptyTableIsZero(t *testing.T) {
+	s := newTestStore(t)
+	ctx := context.Background()
+
+	last, err := s.LastSent(ctx, "service_down")
+	if err != nil {
+		t.Fatalf("last sent on an empty table must not error: %v", err)
+	}
+	if !last.IsZero() {
+		t.Fatalf("want the zero time before anything was sent, got %v", last)
+	}
+}
+
+// TestLastSentIsScopedToItsRule — a send for another rule must not answer for
+// this one, or the repeat interval is clocked off somebody else's alarm.
+func TestLastSentIsScopedToItsRule(t *testing.T) {
+	s := newTestStore(t)
+	ctx := context.Background()
+	at := time.UnixMilli(1_700_000_000_000).UTC()
+
+	if err := s.MarkSent(ctx, "other_rule", at); err != nil {
+		t.Fatalf("mark sent: %v", err)
+	}
+	last, err := s.LastSent(ctx, "service_down")
+	if err != nil {
+		t.Fatalf("last sent: %v", err)
+	}
+	if !last.IsZero() {
+		t.Fatalf("want zero for a rule with no sends, got %v", last)
+	}
+
+	if err := s.MarkSent(ctx, "service_down", at); err != nil {
+		t.Fatalf("mark sent: %v", err)
+	}
+	last, err = s.LastSent(ctx, "service_down")
+	if err != nil {
+		t.Fatalf("last sent: %v", err)
+	}
+	if !last.Equal(at) {
+		t.Fatalf("want %v, got %v", at, last)
+	}
+}

internal/store/digest.go

@@ -3,7 +3,9 @@ package store
 import (
 	"context"
 	"crypto/sha256"
+	"database/sql"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"time"
 )
@@ -44,19 +46,33 @@ func DigestBodyHash(rule, body string) string {
 }
 
 // EnqueueDigestEntry durably records a suppressed care candidate worth
-// resurfacing later. If a pending entry with the same rule+body already
+// resurfacing later. If a LIVE 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.
+//
+// "Live" carries the same expiry test PendingDigestEntries reads with, and for
+// the same reason: a row past its expires_ts is still status='pending' until
+// the sweep gets to it, and the tick enqueues before it sweeps. Deduping
+// against one meant reporting deduped=true against an entry that will never be
+// spoken — the caller drops the phrasing it just paid the LLM for and nothing
+// reaches the bundle. Not yet swept must not mean still deliverable on the
+// write side either.
 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)
+		`SELECT id FROM digest_entries
+		  WHERE status = ? AND rule = ? AND body_hash = ? AND expires_ts > ? LIMIT 1`,
+		DigestPending, rule, hash, now.UnixMilli()).Scan(&existing)
 	if err == nil {
 		return existing, true, nil
 	}
+	if !errors.Is(err, sql.ErrNoRows) {
+		// A real read failure is not "nothing there". Inserting anyway would
+		// duplicate an entry whose existence we never established.
+		return 0, false, fmt.Errorf("enqueue digest entry: dedupe lookup: %w", err)
+	}
 
 	res, err := s.db.ExecContext(ctx,
 		`INSERT INTO digest_entries (rule, severity, body, body_hash, status, created_ts, expires_ts)

internal/store/digest_test.go

@@ -200,3 +200,46 @@ func TestDigestEntryDrainMarksDrainedNotDeleted(t *testing.T) {
 		}
 	}
 }
+
+// TestDigestEnqueueDoesNotDedupeAgainstAnExpiredEntry — an entry past its
+// expires_ts is still status='pending' until the sweep runs, and the tick
+// enqueues before it sweeps. Deduping against one reports deduped=true for a
+// row PendingDigestEntries will never hand back, so the suppressed nudge is
+// dropped instead of held: the entry says the thing was recorded when nothing
+// was.
+func TestDigestEnqueueDoesNotDedupeAgainstAnExpiredEntry(t *testing.T) {
+	s := newTestStore(t)
+	ctx := context.Background()
+	created := time.Now()
+	expiresAt := created.Add(time.Hour)
+
+	first, deduped, err := s.EnqueueDigestEntry(ctx, "break", 2, "ты долго не отдыхала", created, expiresAt)
+	if err != nil {
+		t.Fatalf("enqueue: %v", err)
+	}
+	if deduped {
+		t.Fatal("first enqueue must not report deduped")
+	}
+
+	// A tick after the expiry, with the sweep not yet run: the same suppressed
+	// nudge comes round again and must be recorded afresh.
+	after := expiresAt.Add(time.Minute)
+	second, deduped, err := s.EnqueueDigestEntry(ctx, "break", 2, "ты долго не отдыхала", after, after.Add(time.Hour))
+	if err != nil {
+		t.Fatalf("re-enqueue: %v", err)
+	}
+	if deduped {
+		t.Fatal("an expired entry must not swallow a fresh one")
+	}
+	if second == first {
+		t.Fatalf("want a new row, got the expired one back: id=%d", second)
+	}
+
+	entries, err := s.PendingDigestEntries(ctx, after)
+	if err != nil {
+		t.Fatalf("pending: %v", err)
+	}
+	if len(entries) != 1 || entries[0].ID != second {
+		t.Fatalf("want the fresh entry %d pending, got %+v", second, entries)
+	}
+}