minstrel/internal/sync/compactor.go

// Periodic compactor for the library_changes log. Every mutation in the
// library writes one row, so the log grows without bound unless trimmed. This
// worker trims it daily, keeping each entity's latest row:
//
//   - A row a newer row for the same entity has superseded goes at once. An
//     app reading from any cursor still meets the newer row.
//   - A delete goes once it is CompactorRetention old, and raises the trimmed
//     mark (library_changes_trimmed) past its id. An app whose cursor is below
//     the mark may have missed it, so /api/library/sync answers 410 and the
//     app syncs again from 0.
//   - The latest upsert of an entity that still exists stays, however old. An
//     app syncing from 0 (a fresh install, or after a 410) builds its whole
//     library from the log, so it must hold every artist, album and track.

package sync

import (
	"context"
	"fmt"
	"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 how long a delete stays in the log. An app that
// checks in at least once in that window never misses one; one away longer
// is sent back to 0. Mutable for tests.
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 trims the library_changes log down
// to each entity's latest change.
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 trims the log once and logs what went. Errors are logged at WARN;
// the loop never stops on a transient DB failure.
func (c *Compactor) tickOnce(ctx context.Context) {
	if err := Compact(ctx, c.pool, time.Now().Add(-CompactorRetention), c.logger); err != nil {
		c.logger.Warn("library_changes compactor: trim failed", "err", err)
	}
}

// Compact drops every superseded row, then every delete older than cutoff.
func Compact(ctx context.Context, db dbq.DBTX, cutoff time.Time, logger *slog.Logger) error {
	q := dbq.New(db)
	superseded, err := q.DeleteSupersededLibraryChanges(ctx)
	if err != nil {
		return fmt.Errorf("drop superseded changes: %w", err)
	}
	deletes, err := q.DeleteOldLibraryDeletes(ctx, pgtype.Timestamptz{Time: cutoff, Valid: true})
	if err != nil {
		return fmt.Errorf("drop old deletes: %w", err)
	}
	if superseded > 0 || deletes > 0 {
		logger.Info("library_changes compactor: trimmed rows",
			"superseded", superseded, "old_deletes", deletes, "deletes_older_than", CompactorRetention)
	}
	return nil
}