minstrel/internal/reacquisition/sweeper.go

package reacquisition

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

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

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

// requestCreator is the slice of lidarrrequests.Service the sweeper needs,
// narrowed to an interface so the pass can be tested without a Lidarr client
// or an approval path that talks to one.
type requestCreator interface {
	Create(ctx context.Context, userID pgtype.UUID, p lidarrrequests.CreateParams) (dbq.LidarrRequest, error)
	Approve(ctx context.Context, requestID, adminID pgtype.UUID, ov lidarrrequests.ApproveOverrides) (dbq.LidarrRequest, error)
}

// Sweeper periodically turns albums with long-missing files into Lidarr
// requests (milestone #290).
//
// Deliberately a periodic worker rather than a hook inside the scanner's
// reconcile pass. Reconcile runs inside a scan and has no business deciding
// to talk to a third-party service; it also re-runs often, which would make
// "attempt once, then back off" awkward to express. A worker paces itself,
// survives a restart, and retries without needing another scan.
type Sweeper struct {
	pool     *pgxpool.Pool
	settings *SettingsService
	requests requestCreator
	logger   *slog.Logger
	tick     time.Duration
}

// NewSweeper constructs a Sweeper. The tick is deliberately coarse: the
// smallest meaningful backoff is measured in hours, so waking more often than
// hourly would only re-read settings and find nothing due.
func NewSweeper(
	pool *pgxpool.Pool,
	settings *SettingsService,
	requests requestCreator,
	logger *slog.Logger,
) *Sweeper {
	return &Sweeper{
		pool:     pool,
		settings: settings,
		requests: requests,
		logger:   logger,
		tick:     1 * time.Hour,
	}
}

// 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 {
	Cleared    int64 // albums whose files came back; state dropped
	Requested  int   // requests created this pass
	Approved   int   // of those, sent on to Lidarr
	GaveUp     int   // albums that spent their attempt budget
	Unnameable int64 // albums with missing files but no MBID to ask for
}

// SweepOnce runs one pass. Exported so the admin surface can offer a "run
// now" without waiting out the tick, and 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)

	// Before selecting work: drop state for albums whose files came back, or
	// were adopted at a new path (#2528). Doing this first means a recovered
	// album cannot be picked in the same pass that would have retried it.
	cleared, err := q.ClearRecoveredReacquisitions(ctx)
	if err != nil {
		return fmt.Errorf("clear recovered: %w", err)
	}
	res := PassResult{Cleared: cleared}

	// 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,
		BackoffBaseHours: cfg.BackoffBaseHours,
		BackoffMaxHours:  cfg.BackoffMaxHours,
		PageLimit:        cfg.MaxPerPass,
	})
	if err != nil {
		return fmt.Errorf("list due: %w", err)
	}
	if len(due) == 0 {
		s.logSummary(res)
		return nil
	}

	// One admin lookup per pass, not per album. A library with no admin at
	// all cannot own a request, so the pass stops rather than half-working.
	admin, err := q.GetOldestAdmin(ctx)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			s.logger.Warn("reacquisition: no admin account to own requests; skipping pass")
			return nil
		}
		return fmt.Errorf("owner lookup: %w", err)
	}

	for _, album := range due {
		if err := s.attempt(ctx, q, cfg, admin.ID, album, &res); err != nil {
			// One album's failure must not abandon the rest of the pass: a
			// single unmatched MBID or a transient Lidarr error says nothing
			// about the next album in the list.
			s.logger.Warn("reacquisition: attempt failed",
				"album", album.AlbumTitle, "err", err)
		}
	}
	s.logSummary(res)
	return nil
}

// attempt creates (and optionally approves) the request for one album, then
// records the attempt against its backoff budget.
func (s *Sweeper) attempt(
	ctx context.Context,
	q *dbq.Queries,
	cfg Settings,
	adminID pgtype.UUID,
	album dbq.ListAlbumsDueReacquisitionRow,
	res *PassResult,
) error {
	// The query already filters these out; belt and braces, because Create
	// would reject the request and burn an attempt for no reason.
	if album.AlbumMbid == nil || album.ArtistMbid == nil {
		return nil
	}

	req, err := s.requests.Create(ctx, adminID, lidarrrequests.CreateParams{
		Kind:             "album",
		LidarrArtistMBID: *album.ArtistMbid,
		ArtistName:       album.ArtistName,
		LidarrAlbumMBID:  *album.AlbumMbid,
		AlbumTitle:       album.AlbumTitle,
	})
	if err != nil {
		return fmt.Errorf("create request: %w", err)
	}
	res.Requested++

	// Record the attempt even when Create deduped into somebody else's
	// existing request: the point of the counter is "how often have we gone
	// looking for this album", and a manual request in flight is a reason to
	// wait rather than to keep asking.
	row, err := q.RecordReacquisitionAttempt(ctx, dbq.RecordReacquisitionAttemptParams{
		AlbumID:       album.AlbumID,
		LastRequestID: req.ID,
	})
	if err != nil {
		return fmt.Errorf("record attempt: %w", err)
	}

	if cfg.AutoApprove {
		_, aerr := s.requests.Approve(ctx, req.ID, adminID, lidarrrequests.ApproveOverrides{})
		switch {
		case aerr == nil:
			res.Approved++
		case errors.Is(aerr, lidarrrequests.ErrLidarrDisabled):
			// Leave it pending rather than treating it as a failure. The
			// request is still the right record of intent, and it becomes
			// actionable the moment Lidarr is configured.
			s.logger.Info("reacquisition: request left pending, Lidarr disabled",
				"album", album.AlbumTitle)
		case errors.Is(aerr, lidarrrequests.ErrNotPending):
			// Deduped onto a request somebody already approved. Nothing to do
			// and nothing wrong.
		default:
			return fmt.Errorf("approve: %w", aerr)
		}
	}

	if row.Attempts >= cfg.MaxAttempts {
		if err := q.MarkReacquisitionGaveUp(ctx, album.AlbumID); err != nil {
			return fmt.Errorf("mark gave up: %w", err)
		}
		res.GaveUp++
	}
	return 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.Requested == 0 && res.Cleared == 0 && res.GaveUp == 0 {
		return
	}
	s.logger.Info("reacquisition: sweep",
		"requested", res.Requested,
		"approved", res.Approved,
		"gave_up", res.GaveUp,
		"cleared", res.Cleared,
		"unnameable", res.Unnameable,
	)
}