feat(lidarrrequests): add Reconciler worker matching approved requests to library

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-29 17:17:32 -04:00
parent b11fc117e7
commit f73a5ccef5
2 changed files with 519 additions and 0 deletions
+192
View File
@@ -0,0 +1,192 @@
package lidarrrequests
import (
"context"
"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/lidarrconfig"
)
// Reconciler is a background worker that periodically scans approved
// lidarr_requests and transitions them to completed when their target
// track/album/artist has appeared in the local library (matched by MBID).
// Errors per-row are logged at WARN and do not abort the tick.
type Reconciler struct {
pool *pgxpool.Pool
lidarrCfg *lidarrconfig.Service
logger *slog.Logger
tick time.Duration
batch int32
}
// NewReconciler constructs a Reconciler with production defaults:
// 5-minute tick, batch size 50.
func NewReconciler(pool *pgxpool.Pool, cfg *lidarrconfig.Service, logger *slog.Logger) *Reconciler {
return &Reconciler{
pool: pool,
lidarrCfg: cfg,
logger: logger,
tick: 5 * time.Minute,
batch: 50,
}
}
// Run blocks until ctx is cancelled, ticking every r.tick. Errors from
// tickOnce are logged at WARN and never propagated.
func (r *Reconciler) Run(ctx context.Context) {
t := time.NewTicker(r.tick)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
if err := r.tickOnce(ctx); err != nil {
r.logger.Warn("lidarrrequests: reconciler tick failed", "err", err)
}
}
}
}
// tickOnce loads approved requests (oldest first, up to r.batch) and
// transitions any whose target MBID is now present in the local library.
// It is exported-for-tests via the lowercase name (package-internal).
func (r *Reconciler) tickOnce(ctx context.Context) error {
cfg, err := r.lidarrCfg.Get(ctx)
if err != nil {
return err
}
if !cfg.Enabled {
r.logger.Debug("lidarrrequests: reconciler skipping tick — lidarr disabled")
return nil
}
q := dbq.New(r.pool)
rows, err := q.ListApprovedLidarrRequestsForReconcile(ctx, r.batch)
if err != nil {
return err
}
for _, row := range rows {
if err := r.reconcileRow(ctx, q, row); err != nil {
r.logger.Warn("lidarrrequests: reconciler row failed",
"request_id", row.ID,
"kind", row.Kind,
"err", err,
)
}
}
return nil
}
// reconcileRow checks the library for a match and, if found, calls
// CompleteLidarrRequest. It returns an error only for unexpected DB
// failures; a no-match is not an error.
func (r *Reconciler) reconcileRow(ctx context.Context, q *dbq.Queries, row dbq.LidarrRequest) error {
switch row.Kind {
case dbq.LidarrRequestKindArtist:
return r.reconcileArtist(ctx, q, row)
case dbq.LidarrRequestKindAlbum:
return r.reconcileAlbum(ctx, q, row)
case dbq.LidarrRequestKindTrack:
return r.reconcileTrack(ctx, q, row)
default:
r.logger.Warn("lidarrrequests: reconciler unknown kind", "kind", row.Kind)
return nil
}
}
func (r *Reconciler) reconcileArtist(ctx context.Context, q *dbq.Queries, row dbq.LidarrRequest) error {
var artistID pgtype.UUID
err := r.pool.QueryRow(ctx,
"SELECT id FROM artists WHERE mbid = $1",
row.LidarrArtistMbid,
).Scan(&artistID)
if err != nil {
if isNoRows(err) {
return nil // not in library yet
}
return err
}
_, err = q.CompleteLidarrRequest(ctx, dbq.CompleteLidarrRequestParams{
ID: row.ID,
MatchedArtistID: artistID,
MatchedAlbumID: pgtype.UUID{},
MatchedTrackID: pgtype.UUID{},
})
return err
}
func (r *Reconciler) reconcileAlbum(ctx context.Context, q *dbq.Queries, row dbq.LidarrRequest) error {
if row.LidarrAlbumMbid == nil {
return nil
}
var albumID pgtype.UUID
err := r.pool.QueryRow(ctx,
"SELECT id FROM albums WHERE mbid = $1",
*row.LidarrAlbumMbid,
).Scan(&albumID)
if err != nil {
if isNoRows(err) {
return nil
}
return err
}
_, err = q.CompleteLidarrRequest(ctx, dbq.CompleteLidarrRequestParams{
ID: row.ID,
MatchedAlbumID: albumID,
MatchedArtistID: pgtype.UUID{},
MatchedTrackID: pgtype.UUID{},
})
return err
}
func (r *Reconciler) reconcileTrack(ctx context.Context, q *dbq.Queries, row dbq.LidarrRequest) error {
if row.LidarrAlbumMbid == nil {
return nil
}
// Track-kind requests match via their parent album's MBID, not track.mbid.
var albumID pgtype.UUID
err := r.pool.QueryRow(ctx,
"SELECT id FROM albums WHERE mbid = $1",
*row.LidarrAlbumMbid,
).Scan(&albumID)
if err != nil {
if isNoRows(err) {
return nil
}
return err
}
// Load any track from that album to set matched_track_id.
var trackID pgtype.UUID
err = r.pool.QueryRow(ctx,
"SELECT id FROM tracks WHERE album_id = $1 ORDER BY id LIMIT 1",
albumID,
).Scan(&trackID)
if err != nil {
if isNoRows(err) {
// Album is in the library but no tracks ingested yet — try again next tick.
return nil
}
return err
}
_, err = q.CompleteLidarrRequest(ctx, dbq.CompleteLidarrRequestParams{
ID: row.ID,
MatchedAlbumID: albumID,
MatchedTrackID: trackID,
MatchedArtistID: pgtype.UUID{},
})
return err
}
func isNoRows(err error) bool {
return err == pgx.ErrNoRows
}
@@ -0,0 +1,327 @@
package lidarrrequests
import (
"context"
"io"
"log/slog"
"testing"
"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/lidarrconfig"
)
// seedApprovedRequestDirect inserts a pending request and immediately
// approves it via ApproveLidarrRequest, bypassing the Lidarr API call.
func seedApprovedRequestDirect(t *testing.T, q *dbq.Queries, userID pgtype.UUID, p CreateParams) dbq.LidarrRequest {
t.Helper()
ctx := context.Background()
row, err := q.CreateLidarrRequest(ctx, dbq.CreateLidarrRequestParams{
UserID: userID,
Kind: dbq.LidarrRequestKind(p.Kind),
LidarrArtistMbid: p.LidarrArtistMBID,
LidarrAlbumMbid: nilableStr(p.LidarrAlbumMBID),
LidarrTrackMbid: nilableStr(p.LidarrTrackMBID),
ArtistName: p.ArtistName,
AlbumTitle: nilableStr(p.AlbumTitle),
TrackTitle: nilableStr(p.TrackTitle),
})
if err != nil {
t.Fatalf("seedApprovedRequestDirect create: %v", err)
}
approved, err := q.ApproveLidarrRequest(ctx, dbq.ApproveLidarrRequestParams{
ID: row.ID,
QualityProfileID: nil,
RootFolderPath: nil,
DecidedBy: userID,
})
if err != nil {
t.Fatalf("seedApprovedRequestDirect approve: %v", err)
}
return approved
}
func nilableStr(s string) *string {
if s == "" {
return nil
}
return &s
}
func seedArtist(t *testing.T, q *dbq.Queries, name, mbid string) dbq.Artist {
t.Helper()
a, err := q.UpsertArtist(context.Background(), dbq.UpsertArtistParams{
Name: name, SortName: name, Mbid: &mbid,
})
if err != nil {
t.Fatalf("seedArtist: %v", err)
}
return a
}
func seedAlbum(t *testing.T, q *dbq.Queries, artistID pgtype.UUID, title, mbid string) dbq.Album {
t.Helper()
a, err := q.UpsertAlbum(context.Background(), dbq.UpsertAlbumParams{
Title: title, SortTitle: title,
ArtistID: artistID,
Mbid: &mbid,
})
if err != nil {
t.Fatalf("seedAlbum: %v", err)
}
return a
}
func seedTrack(t *testing.T, q *dbq.Queries, albumID, artistID pgtype.UUID, title, filePath string) dbq.Track {
t.Helper()
tr, err := q.UpsertTrack(context.Background(), dbq.UpsertTrackParams{
Title: title, AlbumID: albumID, ArtistID: artistID,
DurationMs: 180000, FilePath: filePath,
FileSize: 1024, FileFormat: "flac",
})
if err != nil {
t.Fatalf("seedTrack: %v", err)
}
return tr
}
func enableLidarrForPool(t *testing.T, pool *pgxpool.Pool) {
t.Helper()
if err := lidarrconfig.New(pool).Save(context.Background(), lidarrconfig.Config{
Enabled: true, BaseURL: "http://lidarr:8686", APIKey: "test",
}); err != nil {
t.Fatalf("enableLidarrForPool: %v", err)
}
}
func newTestLogger() *slog.Logger {
return slog.New(slog.NewTextHandler(io.Discard, nil))
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
// TestReconciler_MatchesArtistByMBID: an approved artist-kind request whose
// MBID is in the artists table gets completed with matched_artist_id set.
func TestReconciler_MatchesArtistByMBID(t *testing.T) {
pool := newPool(t)
q := dbq.New(pool)
ctx := context.Background()
enableLidarrForPool(t, pool)
user := seedUser(t, pool)
artistMBID := "069b64b6-7884-4f6a-94cc-e4c1d6c87a01"
artist := seedArtist(t, q, "Boards of Canada", artistMBID)
req := seedApprovedRequestDirect(t, q, user, CreateParams{
Kind: "artist", LidarrArtistMBID: artistMBID, ArtistName: "Boards of Canada",
})
rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger())
if err := rec.tickOnce(ctx); err != nil {
t.Fatalf("tickOnce: %v", err)
}
got, err := q.GetLidarrRequestByID(ctx, req.ID)
if err != nil {
t.Fatalf("GetLidarrRequestByID: %v", err)
}
if got.Status != dbq.LidarrRequestStatusCompleted {
t.Errorf("status = %v, want completed", got.Status)
}
if got.MatchedArtistID != artist.ID {
t.Errorf("matched_artist_id = %v, want %v", got.MatchedArtistID, artist.ID)
}
}
// TestReconciler_MatchesAlbumByMBID: an approved album-kind request whose
// album MBID is in the albums table gets completed with matched_album_id set.
func TestReconciler_MatchesAlbumByMBID(t *testing.T) {
pool := newPool(t)
q := dbq.New(pool)
ctx := context.Background()
enableLidarrForPool(t, pool)
user := seedUser(t, pool)
artistMBID := "a-artist-mbid-album-test"
albumMBID := "b-album-mbid-album-test"
artist := seedArtist(t, q, "Test Artist", artistMBID)
album := seedAlbum(t, q, artist.ID, "Test Album", albumMBID)
req := seedApprovedRequestDirect(t, q, user, CreateParams{
Kind: "album", LidarrArtistMBID: artistMBID, ArtistName: "Test Artist",
LidarrAlbumMBID: albumMBID, AlbumTitle: "Test Album",
})
rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger())
if err := rec.tickOnce(ctx); err != nil {
t.Fatalf("tickOnce: %v", err)
}
got, err := q.GetLidarrRequestByID(ctx, req.ID)
if err != nil {
t.Fatalf("GetLidarrRequestByID: %v", err)
}
if got.Status != dbq.LidarrRequestStatusCompleted {
t.Errorf("status = %v, want completed", got.Status)
}
if got.MatchedAlbumID != album.ID {
t.Errorf("matched_album_id = %v, want %v", got.MatchedAlbumID, album.ID)
}
}
// TestReconciler_MatchesTrackViaAlbumMBID: an approved track-kind request
// matches when the parent album appears in the library. matched_track_id is
// also set to a track from that album. The track MBID on the request does
// not need to match any tracks.mbid — matching is via the parent album.
func TestReconciler_MatchesTrackViaAlbumMBID(t *testing.T) {
pool := newPool(t)
q := dbq.New(pool)
ctx := context.Background()
enableLidarrForPool(t, pool)
user := seedUser(t, pool)
artistMBID := "a-artist-mbid-track-test"
albumMBID := "b-album-mbid-track-test"
trackMBID := "c-track-mbid-does-not-exist-in-tracks-table"
artist := seedArtist(t, q, "Track Test Artist", artistMBID)
album := seedAlbum(t, q, artist.ID, "Track Test Album", albumMBID)
track := seedTrack(t, q, album.ID, artist.ID, "Track One", "/music/track-test/01.flac")
req := seedApprovedRequestDirect(t, q, user, CreateParams{
Kind: "track", LidarrArtistMBID: artistMBID, ArtistName: "Track Test Artist",
LidarrAlbumMBID: albumMBID, AlbumTitle: "Track Test Album",
LidarrTrackMBID: trackMBID, TrackTitle: "Track One",
})
rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger())
if err := rec.tickOnce(ctx); err != nil {
t.Fatalf("tickOnce: %v", err)
}
got, err := q.GetLidarrRequestByID(ctx, req.ID)
if err != nil {
t.Fatalf("GetLidarrRequestByID: %v", err)
}
if got.Status != dbq.LidarrRequestStatusCompleted {
t.Errorf("status = %v, want completed", got.Status)
}
if got.MatchedAlbumID != album.ID {
t.Errorf("matched_album_id = %v, want %v", got.MatchedAlbumID, album.ID)
}
if got.MatchedTrackID != track.ID {
t.Errorf("matched_track_id = %v, want %v", got.MatchedTrackID, track.ID)
}
}
// TestReconciler_NoMatchLeavesPending: an approved request whose MBID is not
// in any library table is left unchanged after tickOnce.
func TestReconciler_NoMatchLeavesPending(t *testing.T) {
pool := newPool(t)
q := dbq.New(pool)
ctx := context.Background()
enableLidarrForPool(t, pool)
user := seedUser(t, pool)
req := seedApprovedRequestDirect(t, q, user, CreateParams{
Kind: "artist", LidarrArtistMBID: "nonexistent-mbid-xyz", ArtistName: "Ghost Artist",
})
rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger())
if err := rec.tickOnce(ctx); err != nil {
t.Fatalf("tickOnce: %v", err)
}
got, err := q.GetLidarrRequestByID(ctx, req.ID)
if err != nil {
t.Fatalf("GetLidarrRequestByID: %v", err)
}
if got.Status != dbq.LidarrRequestStatusApproved {
t.Errorf("status = %v, want approved (unchanged)", got.Status)
}
if got.MatchedArtistID.Valid {
t.Errorf("matched_artist_id should be NULL, got %v", got.MatchedArtistID)
}
}
// TestReconciler_AlreadyCompletedRowNotReprocessed: a completed row is not
// touched by the reconciler because ListApprovedLidarrRequestsForReconcile
// filters on status='approved'.
func TestReconciler_AlreadyCompletedRowNotReprocessed(t *testing.T) {
pool := newPool(t)
q := dbq.New(pool)
ctx := context.Background()
enableLidarrForPool(t, pool)
user := seedUser(t, pool)
artistMBID := "already-done-artist-mbid"
artist := seedArtist(t, q, "Already Done Artist", artistMBID)
// Seed approved, then complete it before running tickOnce.
req := seedApprovedRequestDirect(t, q, user, CreateParams{
Kind: "artist", LidarrArtistMBID: artistMBID, ArtistName: "Already Done Artist",
})
completed, err := q.CompleteLidarrRequest(ctx, dbq.CompleteLidarrRequestParams{
ID: req.ID,
MatchedArtistID: artist.ID,
MatchedAlbumID: pgtype.UUID{},
MatchedTrackID: pgtype.UUID{},
})
if err != nil {
t.Fatalf("CompleteLidarrRequest setup: %v", err)
}
rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger())
if err := rec.tickOnce(ctx); err != nil {
t.Fatalf("tickOnce: %v", err)
}
got, err := q.GetLidarrRequestByID(ctx, req.ID)
if err != nil {
t.Fatalf("GetLidarrRequestByID: %v", err)
}
if got.Status != dbq.LidarrRequestStatusCompleted {
t.Errorf("status = %v, want completed (unchanged)", got.Status)
}
if !got.UpdatedAt.Time.Equal(completed.UpdatedAt.Time) {
t.Errorf("updated_at moved: was %v, now %v — reconciler should not have touched completed row",
completed.UpdatedAt.Time, got.UpdatedAt.Time)
}
}
// TestReconciler_DisabledIsNoOp: when lidarr_config.enabled=false the
// reconciler skips the entire tick, leaving approved requests untouched.
func TestReconciler_DisabledIsNoOp(t *testing.T) {
pool := newPool(t)
q := dbq.New(pool)
ctx := context.Background()
// lidarr_config is left disabled (newPool sets it to false).
user := seedUser(t, pool)
artistMBID := "disabled-noop-mbid"
// Seed the matching artist so it WOULD be found if lidarr were enabled.
seedArtist(t, q, "No-Op Artist", artistMBID)
req := seedApprovedRequestDirect(t, q, user, CreateParams{
Kind: "artist", LidarrArtistMBID: artistMBID, ArtistName: "No-Op Artist",
})
rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger())
if err := rec.tickOnce(ctx); err != nil {
t.Fatalf("tickOnce: %v", err)
}
got, err := q.GetLidarrRequestByID(ctx, req.ID)
if err != nil {
t.Fatalf("GetLidarrRequestByID: %v", err)
}
if got.Status != dbq.LidarrRequestStatusApproved {
t.Errorf("status = %v, want approved — reconciler should no-op when disabled", got.Status)
}
}