Maven/internal/store/digest.go

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
}