minstrel/internal/gc/worker.go

// Package gc runs periodic garbage-collection / lifecycle sweeps against
// tables that have no writer-side close path or no retention policy. Each
// sweep is idempotent: re-running it on already-clean rows is a no-op.
//
// One Worker handles all sweeps so a single long-tick goroutine
// amortises the per-tick fixed cost. Each individual sweep is small
// (single UPDATE / DELETE with a time-bounded WHERE) and emits a
// log line with the affected-row count so the sweep cadence is
// visible in the application log without an explicit metrics layer.
//
// Sweeps:
//   - GcCloseStalePlayEvents
//   - GcClosePlaySessionsWithNoRecentEvents
//   - GcExpireScrobbleQueueFailedRows
//   - GcResetStuckSystemPlaylistRuns
//   - GcDeleteExpiredPasswordResets
//   - GcDeleteExpiredSessions          (idle 30d / absolute 1y)
//   - GcPruneDiagnostics               (30d retention)
//   - GcDeleteExpiredSuggestionSnoozes (snoozes expire, then go)
//   - GcDeleteOrphanedCandidateArtistTags(+State) (the similarity feed
//     churns, so cached candidate tags outlive their candidates)
package gc

import (
	"context"
	"log/slog"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
)

// defaultTick is the production sweep cadence. 1 hour is generous
// since each sweep's WHERE clause uses a multi-hour staleness
// threshold; the worst-case delay between a row becoming sweepable
// and the worker noticing is bounded by tick + threshold.
const defaultTick = 1 * time.Hour

// Worker holds the pool + logger + tick interval. Construct with
// [NewWorker]; pass the returned Worker to a goroutine that calls
// [Worker.Run] with a context that's cancelled on shutdown.
type Worker struct {
	pool   *pgxpool.Pool
	logger *slog.Logger
	tick   time.Duration
}

// NewWorker builds a Worker with the production tick (1h). Tests can
// reach into the Worker after construction to override `tick` for
// faster iteration.
func NewWorker(pool *pgxpool.Pool, logger *slog.Logger) *Worker {
	return &Worker{pool: pool, logger: logger, tick: defaultTick}
}

// Run blocks until ctx is cancelled, running every sweep on each
// tick. Sweeps fire in fixed order; an error in one does NOT abort
// the rest (the panic-vs-just-failed distinction matters here — a
// pgx transient error from one query shouldn't prevent the others
// from running).
func (w *Worker) Run(ctx context.Context) {
	// Fire once at start so a freshly-deployed server doesn't wait a
	// full tick before doing the initial sweep. Matches the scrobble
	// + similarity workers' "sweep then tick" pattern.
	w.tickOnce(ctx)
	t := time.NewTicker(w.tick)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			w.tickOnce(ctx)
		}
	}
}

// tickOnce runs each sweep once, logging the affected-row count.
// Errors are logged per-sweep but do NOT abort the remaining ones —
// each sweep is independent.
func (w *Worker) tickOnce(ctx context.Context) {
	q := dbq.New(w.pool)
	w.runSweep(ctx, "close_stale_play_events", q.GcCloseStalePlayEvents)
	w.runSweep(ctx, "close_play_sessions", q.GcClosePlaySessionsWithNoRecentEvents)
	w.runSweep(ctx, "expire_scrobble_failed", q.GcExpireScrobbleQueueFailedRows)
	w.runSweep(ctx, "reset_stuck_system_runs", q.GcResetStuckSystemPlaylistRuns)
	w.runSweep(ctx, "delete_expired_password_resets", q.GcDeleteExpiredPasswordResets)
	w.runSweep(ctx, "delete_expired_sessions", q.GcDeleteExpiredSessions)
	w.runSweep(ctx, "prune_diagnostics", q.GcPruneDiagnostics)
	w.runSweep(ctx, "delete_expired_suggestion_snoozes", q.GcDeleteExpiredSuggestionSnoozes)
	// Tags before state: if the process dies between the two, a candidate left
	// with a state row and no tags simply reads as "settled, nothing found",
	// which is already a valid state. The reverse order could leave tags with
	// no state row, which the drainer would treat as never-processed and
	// re-fetch on top of rows that are already there.
	w.runSweep(ctx, "orphaned_candidate_artist_tags", q.GcDeleteOrphanedCandidateArtistTags)
	w.runSweep(ctx, "orphaned_candidate_artist_tag_state", q.GcDeleteOrphanedCandidateArtistTagState)
}

// runSweep is a small adapter so each sweep call site is a one-liner
// in tickOnce. Logs at info on rows>0 and debug on rows=0 to keep
// the normal-case (nothing-to-do) noise out of operator logs.
func (w *Worker) runSweep(ctx context.Context, name string, fn func(context.Context) (int64, error)) {
	rows, err := fn(ctx)
	if err != nil {
		w.logger.Error("gc sweep failed", "sweep", name, "err", err)
		return
	}
	if rows > 0 {
		w.logger.Info("gc sweep", "sweep", name, "rows_affected", rows)
	} else {
		w.logger.Debug("gc sweep", "sweep", name, "rows_affected", 0)
	}
}