1package storage
 2
 3import (
 4	"context"
 5	"fmt"
 6	"time"
 7)
 8
 9const maxAuditPruneBatch = 100000
10
11var afterAuditPruneBatch = func(int64) {}
12
13func (s *Store) PruneAudit(ctx context.Context, before time.Time, batch int, pause time.Duration) (int64, error) {
14	const operation = "prune audit"
15	if err := s.checkOpen(operation); err != nil {
16		return 0, err
17	}
18	if batch < 1 || batch > maxAuditPruneBatch {
19		return 0, invalid(operation, fmt.Errorf("batch must be between 1 and %d", maxAuditPruneBatch))
20	}
21	var total int64
22	for {
23		n, err := s.pruneAuditBatch(ctx, timeToInt(before), batch)
24		total += n
25		if err != nil {
26			return total, err
27		}
28		afterAuditPruneBatch(n)
29		if n < int64(batch) {
30			return total, nil
31		}
32		if pause > 0 {
33			t := time.NewTimer(pause)
34			select {
35			case <-ctx.Done():
36				t.Stop()
37				return total, ctx.Err()
38			case <-t.C:
39			}
40		} else if err := ctx.Err(); err != nil {
41			return total, err
42		}
43	}
44}
45
46func (s *Store) pruneAuditBatch(ctx context.Context, before int64, batch int) (int64, error) {
47	const operation = "prune audit batch"
48	tx, err := s.sql.BeginTx(ctx, nil)
49	if err != nil {
50		return 0, classify("begin "+operation, err)
51	}
52	defer tx.Rollback()
53	result, err := tx.ExecContext(ctx, `
54DELETE FROM audit_log WHERE id IN (
55    SELECT id FROM audit_log WHERE occurred_at < ? ORDER BY occurred_at, id LIMIT ?
56)`, before, batch)
57	if err != nil {
58		return 0, classify(operation, err)
59	}
60	n, err := result.RowsAffected()
61	if err != nil {
62		return 0, classify(operation, err)
63	}
64	if err := tx.Commit(); err != nil {
65		return 0, classify("commit "+operation, err)
66	}
67	return n, nil
68}