internal/store/retention.go

  1package store
  2
  3import (
  4	"fmt"
  5	"time"
  6)
  7
  8// Nothing was ever deleted from this database. Expired sessions and
  9// tokens were filtered on read and left in place; the audit log, the
 10// activity feed, webhook deliveries and the mail queue grew without
 11// bound. Sweep removes them (#122).
 12
 13// Swept counts what one sweep removed, per table. Zero-valued entries are
 14// left in so a caller logging the result sees every table it asked about.
 15type Swept map[string]int64
 16
 17// Total is how many rows the sweep removed altogether.
 18func (s Swept) Total() int64 {
 19	var n int64
 20	for _, v := range s {
 21		n += v
 22	}
 23	return n
 24}
 25
 26// Retention says how long each capped table keeps a row. A zero duration
 27// means keep forever, which is what an instance that has not configured
 28// retention gets: growing is a decision, but so is deleting an audit
 29// trail, and the second one is not made on an operator's behalf.
 30type Retention struct {
 31	Audit             time.Duration
 32	Events            time.Duration
 33	WebhookDeliveries time.Duration
 34	Mail              time.Duration
 35}
 36
 37// Sweep deletes expired sessions and tokens, then the rows older than
 38// each configured retention. Errors are returned with whatever was
 39// removed before them: a sweep that fails halfway has still done that
 40// much, and the next one picks up the rest.
 41func (s *Store) Sweep(r Retention, now time.Time) (Swept, error) {
 42	out := Swept{}
 43	// Dead the moment they expire, whatever retention says. A used login
 44	// token cannot be replayed and an expired session cannot authenticate,
 45	// so neither is evidence of anything.
 46	expired := []struct {
 47		table string
 48		where string
 49	}{
 50		{"web_sessions", "expires_at <= ?"},
 51		{"login_tokens", "expires_at <= ?"},
 52		{"email_tokens", "expires_at <= ?"},
 53	}
 54	for _, e := range expired {
 55		n, err := s.deleteBy("DELETE FROM "+e.table+" WHERE "+e.where, fmtTime(now))
 56		out[e.table] += n
 57		if err != nil {
 58			return out, fmt.Errorf("sweeping %s: %w", e.table, err)
 59		}
 60	}
 61
 62	// Order matters: deliveries before events. webhook_deliveries.event_id
 63	// is ON DELETE CASCADE, so an event taken out from under a delivery
 64	// takes the delivery with it — including one still queued for retry.
 65	// Sweeping finished deliveries first, and skipping any event that
 66	// still has an unfinished one, keeps that from happening. The effect
 67	// is that a delivery is kept for the shorter of the two retentions,
 68	// which is the honest reading of "keep deliveries for N".
 69	aged := []struct {
 70		table string
 71		where string
 72		keep  time.Duration
 73	}{
 74		{"audit_log", "created_at < ?", r.Audit},
 75		// Only deliveries that have finished: one still being retried is
 76		// live state, however old its first attempt.
 77		{"webhook_deliveries", "created_at < ? AND (delivered_at IS NOT NULL OR failed_at IS NOT NULL)", r.WebhookDeliveries},
 78		{"events", `created_at < ? AND NOT EXISTS (
 79			SELECT 1 FROM webhook_deliveries d
 80			WHERE d.event_id = events.id AND d.delivered_at IS NULL AND d.failed_at IS NULL)`, r.Events},
 81		{"notifications", "created_at < ? AND (sent_at IS NOT NULL OR failed_at IS NOT NULL)", r.Mail},
 82	}
 83	for _, a := range aged {
 84		if a.keep <= 0 {
 85			continue
 86		}
 87		n, err := s.deleteBy("DELETE FROM "+a.table+" WHERE "+a.where, fmtTime(now.Add(-a.keep)))
 88		out[a.table] += n
 89		if err != nil {
 90			return out, fmt.Errorf("sweeping %s: %w", a.table, err)
 91		}
 92	}
 93	return out, nil
 94}
 95
 96func (s *Store) deleteBy(q string, args ...any) (int64, error) {
 97	res, err := s.DB.Exec(q, args...)
 98	if err != nil {
 99		return 0, err
100	}
101	n, _ := res.RowsAffected()
102	return n, nil
103}