package store import ( "context" "crypto/sha256" "database/sql" "encoding/hex" "errors" "fmt" "time" ) // DigestStatus — the lifecycle state of a digest entry. Go has no enum type; // the idiom is a defined type plus constants, which is what this is. The point // is not ceremony: with bare strings nothing stopped a rule name or a body // hash being passed where a status belongs, and every one of these values // reaches SQL. A defined type makes that a compile error. type DigestStatus string // 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 DigestStatus = "pending" DigestDrained DigestStatus = "drained" DigestExpired DigestStatus = "expired" ) // DigestEntry — one gate-suppressed care candidate durably held for later // bundled delivery. type DigestEntry struct { ID int64 Rule string Severity int Body string CandidateFingerprint string CreatedTs time.Time ExpiresTs time.Time } // DigestBodyHash records the exact presentation stored in a digest entry. The // pre-phrase dedupe key is CandidateFingerprint; body_hash remains useful for // audit/integrity and for legacy rows written before candidate identity was // persisted. func DigestBodyHash(rule, body string) string { sum := sha256.Sum256([]byte(rule + "\x00" + body)) return hex.EncodeToString(sum[:8]) } // LiveDigestEntry returns the pending, unexpired entry for one semantic // candidate occurrence. This is intentionally a store read rather than an // in-memory cache: the caller uses it before PhraseNudge, including on the // first tick after a daemon restart. func (s *Store) LiveDigestEntry(ctx context.Context, rule, candidateFingerprint string, now time.Time) (DigestEntry, bool, error) { if candidateFingerprint == "" { return DigestEntry{}, false, errors.New("live digest entry: empty candidate fingerprint") } row := s.db.QueryRowContext(ctx, `SELECT id, rule, severity, body, candidate_fingerprint, created_ts, expires_ts FROM digest_entries WHERE status = ? AND rule = ? AND candidate_fingerprint = ? AND expires_ts > ? LIMIT 1`, DigestPending, rule, candidateFingerprint, now.UnixMilli()) entry, err := scanDigestEntry(row) if errors.Is(err, sql.ErrNoRows) { return DigestEntry{}, false, nil } if err != nil { return DigestEntry{}, false, fmt.Errorf("live digest entry: %w", err) } return entry, true, nil } // EnqueueDigestEntry durably records a suppressed care candidate worth // resurfacing later. If a LIVE pending entry with the same rule+candidate // fingerprint already exists, this is a no-op that returns the existing id // and deduped=true. The lookup and insert share a transaction so a second // caller cannot race the pre-phrase read into a duplicate row. // // "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. A matching stale row is expired inside this transaction // before insertion so the partial unique index does not make sweep order part // of correctness. func (s *Store) EnqueueDigestEntry(ctx context.Context, rule, candidateFingerprint string, severity int, body string, now, expiresAt time.Time) (id int64, deduped bool, err error) { if candidateFingerprint == "" { return 0, false, errors.New("enqueue digest entry: empty candidate fingerprint") } hash := DigestBodyHash(rule, body) tx, err := s.db.BeginTx(ctx, nil) if err != nil { return 0, false, fmt.Errorf("enqueue digest entry: begin: %w", err) } defer func() { _ = tx.Rollback() }() if _, err := tx.ExecContext(ctx, `UPDATE digest_entries SET status = ? WHERE status = ? AND rule = ? AND candidate_fingerprint = ? AND expires_ts <= ?`, DigestExpired, DigestPending, rule, candidateFingerprint, now.UnixMilli()); err != nil { return 0, false, fmt.Errorf("enqueue digest entry: expire stale candidate: %w", err) } var existing int64 err = tx.QueryRowContext(ctx, `SELECT id FROM digest_entries WHERE status = ? AND rule = ? AND candidate_fingerprint = ? AND expires_ts > ? LIMIT 1`, DigestPending, rule, candidateFingerprint, now.UnixMilli()).Scan(&existing) if err == nil { if err := tx.Commit(); err != nil { return 0, false, fmt.Errorf("enqueue digest entry: dedupe commit: %w", err) } 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 := tx.ExecContext(ctx, `INSERT INTO digest_entries (rule, severity, body, body_hash, candidate_fingerprint, status, created_ts, expires_ts) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, rule, severity, body, hash, candidateFingerprint, 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) } if err := tx.Commit(); err != nil { return 0, false, fmt.Errorf("enqueue digest entry: commit: %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, candidate_fingerprint, 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() { e, err := scanDigestEntry(rows) if err != nil { return nil, fmt.Errorf("pending digest entries: scan: %w", err) } out = append(out, e) } return out, rows.Err() } type digestScanner interface { Scan(dest ...any) error } func scanDigestEntry(row digestScanner) (DigestEntry, error) { var e DigestEntry var created, expires int64 if err := row.Scan(&e.ID, &e.Rule, &e.Severity, &e.Body, &e.CandidateFingerprint, &created, &expires); err != nil { return DigestEntry{}, err } e.CreatedTs = time.UnixMilli(created) e.ExpiresTs = time.UnixMilli(expires) return e, nil } // 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 }