package store import ( "context" "database/sql" "fmt" "time" ) // Retention windows. Each must comfortably exceed every retry and recovery // horizon that could still consult the row, because deleting one early changes // behaviour rather than merely reclaiming space: // // - An idempotency key must outlive any client retry of the same mutation; // deleting it early turns a replay into a second real mutation. The client // heartbeats every 10s and abandons a ticket far sooner than this. // - A published outbox row is only kept for operator forensics; delivery has // already happened, and the dispatcher never re-reads a published row. // - A session row past expiry can no longer authenticate, so retaining it // buys nothing beyond a short audit tail. const ( IdempotencyKeyRetention = 24 * time.Hour PublishedOutboxRetention = 72 * time.Hour ExpiredSessionRetention = 24 * time.Hour // DeadLetteredOutboxRetention is deliberately the longest: those rows are // the record of events that were never delivered, and an operator needs // time to notice and investigate them. DeadLetteredOutboxRetention = 30 * 24 * time.Hour ) // Deletes are batched and use SKIP LOCKED so a purge never blocks live // traffic, never holds a long transaction, and multiple maintenance replicas // can run concurrently without contending on the same rows. const ( PurgeIdempotencyKeysSQL = `DELETE FROM idempotency_keys WHERE (scope, idempotency_key) IN ( SELECT scope, idempotency_key FROM idempotency_keys WHERE created_at < $1 ORDER BY created_at LIMIT $2 FOR UPDATE SKIP LOCKED )` PurgePublishedOutboxSQL = `DELETE FROM outbox WHERE event_id IN ( SELECT event_id FROM outbox WHERE published_at IS NOT NULL AND published_at < $1 ORDER BY published_at LIMIT $2 FOR UPDATE SKIP LOCKED )` PurgeDeadLetteredOutboxSQL = `DELETE FROM outbox WHERE event_id IN ( SELECT event_id FROM outbox WHERE dead_lettered_at IS NOT NULL AND dead_lettered_at < $1 ORDER BY dead_lettered_at LIMIT $2 FOR UPDATE SKIP LOCKED )` PurgeExpiredSessionsSQL = `DELETE FROM sessions WHERE session_id IN ( SELECT session_id FROM sessions WHERE expires_at < $1 ORDER BY expires_at LIMIT $2 FOR UPDATE SKIP LOCKED )` ) // RetentionReport is the per-pass result. Callers expose these as metrics so // deletion lag is observable: if a count stays pinned at the batch size, the // purge is not keeping up with insert volume. type RetentionReport struct { IdempotencyKeys int64 PublishedOutbox int64 DeadLetteredOutbox int64 ExpiredSessions int64 } func (r RetentionReport) Total() int64 { return r.IdempotencyKeys + r.PublishedOutbox + r.DeadLetteredOutbox + r.ExpiredSessions } // PurgeExpiredRecords removes one bounded batch from each retained table. It // returns partial progress alongside an error so a failure in one table does // not hide the work already done in another. func PurgeExpiredRecords(ctx context.Context, db *sql.DB, now time.Time, batch int) (RetentionReport, error) { var report RetentionReport if db == nil || now.IsZero() || batch < 1 || batch > 10000 { return report, fmt.Errorf("invalid retention arguments") } steps := []struct { query string cutoff time.Time into *int64 }{ {PurgeIdempotencyKeysSQL, now.Add(-IdempotencyKeyRetention), &report.IdempotencyKeys}, {PurgePublishedOutboxSQL, now.Add(-PublishedOutboxRetention), &report.PublishedOutbox}, {PurgeDeadLetteredOutboxSQL, now.Add(-DeadLetteredOutboxRetention), &report.DeadLetteredOutbox}, {PurgeExpiredSessionsSQL, now.Add(-ExpiredSessionRetention), &report.ExpiredSessions}, } for _, step := range steps { result, err := db.ExecContext(ctx, step.query, step.cutoff, batch) if err != nil { return report, err } deleted, err := result.RowsAffected() if err != nil { return report, err } *step.into = deleted } return report, nil } // RetentionBacklog counts rows already past their retention window. This is // the deletion-lag metric: a number that keeps climbing means the purge // interval or batch size is too small for current volume. func RetentionBacklog(ctx context.Context, db *sql.DB, now time.Time) (int64, error) { if db == nil || now.IsZero() { return 0, fmt.Errorf("invalid retention backlog arguments") } const query = `SELECT (SELECT count(*) FROM idempotency_keys WHERE created_at < $1) + (SELECT count(*) FROM outbox WHERE published_at IS NOT NULL AND published_at < $2) + (SELECT count(*) FROM sessions WHERE expires_at < $3)` var backlog int64 err := db.QueryRowContext(ctx, query, now.Add(-IdempotencyKeyRetention), now.Add(-PublishedOutboxRetention), now.Add(-ExpiredSessionRetention), ).Scan(&backlog) return backlog, err }