minstrel/internal/catalogue/catalogue.go

// Package catalogue is Minstrel's own record of what artists have released,
// read from MusicBrainz and kept in the database (M502 #5519). The wanted
// list, search and import read it; MusicBrainz is asked only when what is
// held is missing or old.
package catalogue

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/musicbrainz"
)

const (
	// DiscographyMaxAge is how long an artist's list of release groups is
	// trusted before it is read again. New releases appear on this cadence.
	DiscographyMaxAge = 7 * 24 * time.Hour
	// ReleasesMaxAge is how long a release group's editions and track lists
	// are trusted. They change rarely once a group is out.
	ReleasesMaxAge = 30 * 24 * time.Hour
)

// ErrNotFound: MusicBrainz has no artist or release group under that id.
var ErrNotFound = errors.New("catalogue: not in MusicBrainz")

// Service reads and refreshes the catalogue.
type Service struct {
	pool   *pgxpool.Pool
	mb     *musicbrainz.Client
	logger *slog.Logger
	now    func() time.Time
}

// New makes a Service. A nil client talks to the public MusicBrainz.
func New(pool *pgxpool.Pool, mb *musicbrainz.Client, logger *slog.Logger) *Service {
	if mb == nil {
		mb = &musicbrainz.Client{}
	}
	if logger == nil {
		logger = slog.Default()
	}
	return &Service{pool: pool, mb: mb, logger: logger, now: time.Now}
}

// MusicBrainz is the client the service asks, for callers that search.
func (s *Service) MusicBrainz() *musicbrainz.Client { return s.mb }

// Discography is an artist and the release groups credited to them.
type Discography struct {
	Artist dbq.MbArtist
	Groups []dbq.MbReleaseGroup
	// Stale: MusicBrainz could not be asked, so this is the last copy read.
	Stale bool
}

// Discography returns the artist's release groups, reading them from
// MusicBrainz first when they have never been read or are older than
// DiscographyMaxAge. When MusicBrainz is unreachable, a copy already held is
// returned marked Stale; with nothing held, the error is returned.
func (s *Service) Discography(ctx context.Context, artistMBID string) (Discography, error) {
	return s.DiscographyWithin(ctx, artistMBID, DiscographyMaxAge)
}

// DiscographyWithin is Discography with the age past which the held copy is
// read again: a followed artist is checked for new releases more often than
// one the catalogue merely knows.
func (s *Service) DiscographyWithin(ctx context.Context, artistMBID string, maxAge time.Duration) (Discography, error) {
	q := dbq.New(s.pool)
	artist, err := q.GetMBArtist(ctx, artistMBID)
	held := err == nil && artist.DiscographyRefreshedAt.Valid
	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return Discography{}, fmt.Errorf("catalogue: read artist: %w", err)
	}
	stale := false
	if !held || s.now().Sub(artist.DiscographyRefreshedAt.Time) > maxAge {
		if rerr := s.RefreshArtist(ctx, artistMBID); rerr != nil {
			if !held || !errors.Is(rerr, musicbrainz.ErrUnreachable) {
				return Discography{}, rerr
			}
			stale = true
		}
		if artist, err = q.GetMBArtist(ctx, artistMBID); err != nil {
			return Discography{}, fmt.Errorf("catalogue: read artist: %w", err)
		}
	}
	groups, err := q.ListMBArtistReleaseGroups(ctx, artistMBID)
	if err != nil {
		return Discography{}, fmt.Errorf("catalogue: read discography: %w", err)
	}
	return Discography{Artist: artist, Groups: groups, Stale: stale}, nil
}

// RefreshArtist reads the artist and every release group credited to them,
// and replaces the held discography in one transaction.
func (s *Service) RefreshArtist(ctx context.Context, artistMBID string) error {
	artist, err := s.mb.Artist(ctx, artistMBID)
	if err != nil {
		return mbErr(err)
	}
	groups, err := s.mb.ArtistReleaseGroups(ctx, artistMBID)
	if err != nil {
		return mbErr(err)
	}
	return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
		q := dbq.New(tx)
		if err := q.UpsertMBArtist(ctx, dbq.UpsertMBArtistParams{
			Mbid: artist.MBID, Name: artist.Name, SortName: artist.SortName,
			Disambiguation: artist.Disambiguation, ArtistType: artist.Type, Country: artist.Country,
		}); err != nil {
			return fmt.Errorf("catalogue: save artist: %w", err)
		}
		if err := q.DeleteMBArtistReleaseGroups(ctx, artist.MBID); err != nil {
			return fmt.Errorf("catalogue: clear discography: %w", err)
		}
		for _, g := range groups {
			if err := upsertGroup(ctx, q, g); err != nil {
				return err
			}
			if err := q.AddMBArtistReleaseGroup(ctx, dbq.AddMBArtistReleaseGroupParams{
				ArtistMbid: artist.MBID, ReleaseGroupMbid: g.MBID,
			}); err != nil {
				return fmt.Errorf("catalogue: link release group: %w", err)
			}
		}
		return q.MarkMBArtistDiscographyRefreshed(ctx, artist.MBID)
	})
}

func upsertGroup(ctx context.Context, q *dbq.Queries, g musicbrainz.ReleaseGroup) error {
	secondary := g.SecondaryTypes
	if secondary == nil {
		secondary = []string{}
	}
	if err := q.UpsertMBReleaseGroup(ctx, dbq.UpsertMBReleaseGroupParams{
		Mbid: g.MBID, Title: g.Title, PrimaryType: g.PrimaryType, SecondaryTypes: secondary,
		FirstReleaseDate: g.FirstReleaseDate, ArtistCredit: g.ArtistCredit,
	}); err != nil {
		return fmt.Errorf("catalogue: save release group: %w", err)
	}
	return nil
}

// Edition is one release of a group with its track list.
type Edition struct {
	Release dbq.MbRelease
	Tracks  []dbq.MbReleaseTrack
}

// Editions is a release group and its releases.
type Editions struct {
	Group    dbq.MbReleaseGroup
	Editions []Edition
	Stale    bool
}

// Releases returns a release group's releases with their track lists, read
// from MusicBrainz first when never read or older than ReleasesMaxAge. Like
// Discography, an unreachable MusicBrainz serves the held copy as Stale.
func (s *Service) Releases(ctx context.Context, groupMBID string) (Editions, error) {
	q := dbq.New(s.pool)
	group, err := q.GetMBReleaseGroup(ctx, groupMBID)
	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return Editions{}, fmt.Errorf("catalogue: read release group: %w", err)
	}
	held := err == nil && group.ReleasesRefreshedAt.Valid
	stale := false
	if !held || s.now().Sub(group.ReleasesRefreshedAt.Time) > ReleasesMaxAge {
		if rerr := s.RefreshReleaseGroup(ctx, groupMBID); rerr != nil {
			if !held || !errors.Is(rerr, musicbrainz.ErrUnreachable) {
				return Editions{}, rerr
			}
			stale = true
		}
		if group, err = q.GetMBReleaseGroup(ctx, groupMBID); err != nil {
			return Editions{}, fmt.Errorf("catalogue: read release group: %w", err)
		}
	}
	releases, err := q.ListMBReleasesForGroup(ctx, groupMBID)
	if err != nil {
		return Editions{}, fmt.Errorf("catalogue: read releases: %w", err)
	}
	tracks, err := q.ListMBReleaseTracksForGroup(ctx, groupMBID)
	if err != nil {
		return Editions{}, fmt.Errorf("catalogue: read track lists: %w", err)
	}
	byRelease := make(map[string][]dbq.MbReleaseTrack, len(releases))
	for _, t := range tracks {
		byRelease[t.ReleaseMbid] = append(byRelease[t.ReleaseMbid], t)
	}
	out := Editions{Group: group, Stale: stale}
	for _, r := range releases {
		out.Editions = append(out.Editions, Edition{Release: r, Tracks: byRelease[r.Mbid]})
	}
	return out, nil
}

// RefreshReleaseGroup reads a group's releases and track lists and replaces
// the held copy in one transaction. A group not yet held is looked up first.
func (s *Service) RefreshReleaseGroup(ctx context.Context, groupMBID string) error {
	q := dbq.New(s.pool)
	var group *musicbrainz.ReleaseGroup
	if _, err := q.GetMBReleaseGroup(ctx, groupMBID); errors.Is(err, pgx.ErrNoRows) {
		g, gerr := s.mb.ReleaseGroup(ctx, groupMBID)
		if gerr != nil {
			return mbErr(gerr)
		}
		group = &g
	} else if err != nil {
		return fmt.Errorf("catalogue: read release group: %w", err)
	}
	releases, err := s.mb.ReleaseGroupReleases(ctx, groupMBID)
	if err != nil {
		return mbErr(err)
	}
	return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
		tq := dbq.New(tx)
		if group != nil {
			if err := upsertGroup(ctx, tq, *group); err != nil {
				return err
			}
		}
		if err := tq.DeleteMBReleasesForGroup(ctx, groupMBID); err != nil {
			return fmt.Errorf("catalogue: clear releases: %w", err)
		}
		for _, r := range releases {
			if err := tq.InsertMBRelease(ctx, dbq.InsertMBReleaseParams{
				Mbid: r.MBID, ReleaseGroupMbid: groupMBID, Title: r.Title, Status: r.Status,
				ReleaseDate: r.Date, Country: r.Country, Disambiguation: r.Disambiguation,
				Formats: r.Formats(), TrackCount: int32(r.TrackCount()),
			}); err != nil {
				return fmt.Errorf("catalogue: save release: %w", err)
			}
			for _, m := range r.Media {
				for _, t := range m.Tracks {
					if err := tq.InsertMBReleaseTrack(ctx, dbq.InsertMBReleaseTrackParams{
						ReleaseMbid: r.MBID, MediumPosition: int32(m.Position), Position: int32(t.Position),
						MediumFormat: m.Format, Number: t.Number, Title: t.Title,
						LengthMs: int32(t.LengthMs), RecordingMbid: t.RecordingMBID,
					}); err != nil {
						return fmt.Errorf("catalogue: save track: %w", err)
					}
				}
			}
		}
		return tq.MarkMBReleaseGroupReleasesRefreshed(ctx, groupMBID)
	})
}

// RefreshDue re-reads up to limit discographies that are older than
// DiscographyMaxAge, never-read first. It stops at the first unreachable
// MusicBrainz, since the rest would fail the same way.
func (s *Service) RefreshDue(ctx context.Context, limit int) (refreshed int, err error) {
	due, err := dbq.New(s.pool).ListMBArtistsDueForRefresh(ctx, dbq.ListMBArtistsDueForRefreshParams{
		Before:  pgtype.Timestamptz{Time: s.now().Add(-DiscographyMaxAge), Valid: true},
		MaxRows: int32(limit),
	})
	if err != nil {
		return 0, fmt.Errorf("catalogue: list due artists: %w", err)
	}
	for _, mbid := range due {
		if err := s.RefreshArtist(ctx, mbid); err != nil {
			if errors.Is(err, musicbrainz.ErrUnreachable) || ctx.Err() != nil {
				return refreshed, err
			}
			// One artist MusicBrainz cannot describe (merged away, say)
			// must not stop the rest.
			s.logger.Warn("catalogue: refresh artist", "mbid", mbid, "err", err)
			continue
		}
		refreshed++
	}
	return refreshed, nil
}

// RefreshInterval is how often Run looks for discographies due a re-read,
// and RefreshBatch how many it reads per look: 20 artists is about a minute
// of MusicBrainz's budget.
const (
	RefreshInterval = time.Hour
	RefreshBatch    = 20
)

// Run refreshes due discographies every RefreshInterval until ctx ends.
func (s *Service) Run(ctx context.Context) {
	ticker := time.NewTicker(RefreshInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
		n, err := s.RefreshDue(ctx, RefreshBatch)
		switch {
		case err != nil && ctx.Err() == nil:
			s.logger.Warn("catalogue: refresh pass stopped", "refreshed", n, "err", err)
		case n > 0:
			s.logger.Info("catalogue: refresh pass complete", "refreshed", n)
		}
	}
}

// mbErr maps MusicBrainz's not-found to the catalogue's; anything else passes
// through so callers can tell an unreachable MusicBrainz apart.
func mbErr(err error) error {
	if errors.Is(err, musicbrainz.ErrNotFound) {
		return fmt.Errorf("%w: %v", ErrNotFound, err)
	}
	return err
}