minstrel/internal/reacquisition/sweeper.go

package reacquisition

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

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

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

// Sweeper periodically wants back albums whose files have been missing past
// the grace window (milestone #290, M502 #5520). Searching for them is the
// search schedule's (#5521), which retries per want.
//
// Deliberately a periodic worker rather than a hook inside the scanner's
// reconcile pass: reconcile runs inside a scan and re-runs often, and a
// filesystem lies transiently (an unmounted volume, a NAS mid-reboot) in ways
// the grace window exists to ride out.
type Sweeper struct {
	pool     *pgxpool.Pool
	settings *SettingsService
	logger   *slog.Logger
	tick     time.Duration
	// releaseGroup names the MusicBrainz release group of a release id, for an
	// album whose tags never carried one (#5241). Wants are kept by release
	// group. A field so tests can stand in for MusicBrainz.
	releaseGroup func(ctx context.Context, releaseMBID string) (string, error)
}

// NewSweeper constructs a Sweeper. The tick is deliberately coarse: the grace
// window is measured in hours.
func NewSweeper(pool *pgxpool.Pool, settings *SettingsService, logger *slog.Logger) *Sweeper {
	return &Sweeper{
		pool:         pool,
		settings:     settings,
		logger:       logger,
		tick:         1 * time.Hour,
		releaseGroup: tags.ReleaseGroupForRelease,
	}
}

// Run drives the sweep loop until ctx is cancelled.
func (s *Sweeper) Run(ctx context.Context) {
	t := time.NewTicker(s.tick)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			if err := s.SweepOnce(ctx); err != nil {
				s.logger.Warn("reacquisition: sweep failed", "err", err)
			}
		}
	}
}

// PassResult reports what a single sweep did, for logging and tests.
type PassResult struct {
	Wanted     int   // albums newly wanted, or wanted again after being held
	Unnameable int64 // albums with missing files but no MBID to want them by
	// Unresolved: albums skipped this pass because MusicBrainz could not name
	// their release group; the next pass tries again.
	Unresolved int
}

// SweepOnce runs one pass. Exported so tests drive it directly.
func (s *Sweeper) SweepOnce(ctx context.Context) error {
	cfg := s.settings.Get()
	if !cfg.Enabled {
		return nil
	}
	q := dbq.New(s.pool)
	var res PassResult
	// Counted, not acted on: an album MusicBrainz cannot name is not a
	// failure to retry, it is a permanent gap the operator should see.
	if n, cerr := q.CountAlbumsMissingWithoutMbid(ctx); cerr == nil {
		res.Unnameable = n
	}
	due, err := q.ListAlbumsDueReacquisition(ctx, dbq.ListAlbumsDueReacquisitionParams{
		GraceHours: cfg.GraceHours,
		PageLimit:  cfg.MaxPerPass,
	})
	if err != nil {
		return fmt.Errorf("list due: %w", err)
	}
	for _, album := range due {
		if err := s.want(ctx, q, album, &res); err != nil {
			// One album's failure must not abandon the rest of the pass.
			s.logger.Warn("reacquisition: want failed", "album", album.AlbumTitle, "err", err)
		}
	}
	s.logSummary(res)
	return nil
}

// want records one album as wanted under its release group.
func (s *Sweeper) want(ctx context.Context, q *dbq.Queries, album dbq.ListAlbumsDueReacquisitionRow, res *PassResult) error {
	// The query already filters these out; belt and braces.
	if album.AlbumMbid == nil || album.ArtistMbid == nil {
		return nil
	}
	group, err := s.albumReleaseGroup(ctx, q, album)
	if err != nil {
		if errors.Is(err, tags.ErrNotFound) {
			res.Unresolved++
			s.logger.Info("reacquisition: no MusicBrainz release group for album; skipped",
				"album", album.AlbumTitle, "release_mbid", *album.AlbumMbid)
			return nil
		}
		return fmt.Errorf("release group: %w", err)
	}
	_, err = q.WantMissingAlbum(ctx, dbq.WantMissingAlbumParams{
		ReleaseGroupMbid: group, ArtistMbid: *album.ArtistMbid, ArtistName: album.ArtistName,
		Title: album.AlbumTitle, AlbumID: album.AlbumID,
	})
	if errors.Is(err, pgx.ErrNoRows) {
		// Already wanted, open: by a request, a follow, or this sweeper.
		return nil
	}
	if err != nil {
		return fmt.Errorf("want: %w", err)
	}
	res.Wanted++
	return nil
}

// albumReleaseGroup is the album's release group: the one its tags carried,
// else MusicBrainz's answer for its release id, cached onto the album so the
// next pass need not ask again. Returns tags.ErrNotFound when MusicBrainz
// cannot name it.
func (s *Sweeper) albumReleaseGroup(ctx context.Context, q *dbq.Queries, album dbq.ListAlbumsDueReacquisitionRow) (string, error) {
	if album.AlbumReleaseGroupMbid != nil && *album.AlbumReleaseGroupMbid != "" {
		return *album.AlbumReleaseGroupMbid, nil
	}
	group, err := s.releaseGroup(ctx, *album.AlbumMbid)
	if err != nil {
		return "", err
	}
	if cerr := q.SetAlbumReleaseGroupMbidIfNull(ctx, dbq.SetAlbumReleaseGroupMbidIfNullParams{
		ID:               album.AlbumID,
		ReleaseGroupMbid: &group,
	}); cerr != nil {
		s.logger.Warn("reacquisition: cache release group failed", "album", album.AlbumTitle, "err", cerr)
	}
	return group, nil
}

func (s *Sweeper) logSummary(res PassResult) {
	// Silence when a pass did nothing at all — this runs hourly forever, and
	// an unconditional line would bury the passes that mattered.
	if res.Wanted == 0 && res.Unresolved == 0 {
		return
	}
	s.logger.Info("reacquisition: sweep",
		"wanted", res.Wanted,
		"unnameable", res.Unnameable,
		"unresolved", res.Unresolved,
	)
}