// Periodic compactor for the library_changes log. Every mutation in the // library writes one row, so the table grows unbounded without a retention // policy. This worker trims rows older than CompactorRetention on a daily // tick. // // Why 30 days: clients keep a cursor and pull deltas from the last seen id. // A client that checks in at least once per retention window never sees its // cursor go stale; one offline longer than that gets the /api/library/sync // 410 and does a full resync. package sync import ( "context" "log/slog" "time" "github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgxpool" "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" ) // CompactorRetention is the age past which library_changes rows are // dropped. Mutable for tests; production runs should not change it // without considering the corresponding client cursor-stale fallback. var CompactorRetention = 30 * 24 * time.Hour // CompactorInterval is how often the worker checks. Daily is enough — // the table grows on the order of a few thousand rows/day for an // active library, well below a size where intra-day trimming matters. var CompactorInterval = 24 * time.Hour // Compactor is a background worker that periodically deletes // library_changes rows older than CompactorRetention. type Compactor struct { pool *pgxpool.Pool logger *slog.Logger interval time.Duration } // NewCompactor builds a compactor with production defaults. func NewCompactor(pool *pgxpool.Pool, logger *slog.Logger) *Compactor { return &Compactor{ pool: pool, logger: logger, interval: CompactorInterval, } } // Run blocks until ctx is cancelled, ticking every c.interval. // Fires one tick immediately on startup so a long-running process // that hasn't been compacted in a while catches up. func (c *Compactor) Run(ctx context.Context) { c.tickOnce(ctx) t := time.NewTicker(c.interval) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: c.tickOnce(ctx) } } } // tickOnce deletes rows older than CompactorRetention and logs the // row count. Errors are logged at WARN; we never abort the loop on // transient DB failures. func (c *Compactor) tickOnce(ctx context.Context) { cutoff := pgtype.Timestamptz{Time: time.Now().Add(-CompactorRetention), Valid: true} rows, err := dbq.New(c.pool).DeleteOldLibraryChanges(ctx, cutoff) if err != nil { c.logger.Warn("library_changes compactor: delete failed", "err", err) return } if rows > 0 { c.logger.Info("library_changes compactor: trimmed rows", "count", rows, "older_than", CompactorRetention) } }