minstrel/internal/importer/importer.go

package importer

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"slices"
	"strconv"
	"sync"
	"time"

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

	"git.fabledsword.com/bvandeusen/minstrel/internal/acquire"
	"git.fabledsword.com/bvandeusen/minstrel/internal/catalogue"
	"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
	"git.fabledsword.com/bvandeusen/minstrel/internal/downloader"
	"git.fabledsword.com/bvandeusen/minstrel/internal/indexer"
)

// Import states, as grabs.import_state holds them.
const (
	StateDownloading = "downloading"
	StateWaiting     = "waiting"
	StateImporting   = "importing"
	StateImported    = "imported"
	StateRefused     = "refused"
	StateReview      = "review"
	StateHeld        = "held"
	StateFailed      = "failed"
)

const (
	// interval: a finished download is noticed within a minute. One
	// Torrents call per download client, so the client barely notices.
	interval = time.Minute
	// passTimeout bounds a pass: fingerprinting a long album against
	// AcoustID's pace takes a minute or two, so half an hour is a hang.
	passTimeout = 30 * time.Minute
	// staleMinutes: an import still "importing" after this was cut short
	// (a restart mid-copy) and is tried again.
	staleMinutes = 30
)

// Scanner reads new files into the library.
type Scanner interface {
	ScanFiles(ctx context.Context, paths []string) ([]pgtype.UUID, error)
}

// Catalogue lists a release group's editions.
type Catalogue interface {
	Releases(ctx context.Context, groupMBID string) (catalogue.Editions, error)
}

// Importer watches every live grab's download and imports it once finished.
type Importer struct {
	pool      *pgxpool.Pool
	Scanner   Scanner
	Catalogue Catalogue
	Printer   Printer
	// Roots are the library's folders (the scan paths). New music is filed
	// under the one the settings name, or the first.
	Roots []string
	// DataDir is where cached art lives, for tidying an artist an upgrade
	// leaves with nothing (#5524). Empty skips it.
	DataDir string
	// ClientFor builds the download client a grab went to. Defaults to
	// acquire.DownloadClientFor.
	ClientFor func(ctx context.Context, code string) (downloader.Driver, error)
	logger    *slog.Logger
	wake      chan struct{}
	mu        sync.Mutex // one pass at a time
}

// New returns an Importer. printer may be nil: the check then rests on
// lengths and tags.
func New(pool *pgxpool.Pool, scanner Scanner, cat Catalogue, printer Printer, roots []string, logger *slog.Logger) *Importer {
	if logger == nil {
		logger = slog.Default()
	}
	im := &Importer{pool: pool, Scanner: scanner, Catalogue: cat, Printer: printer, Roots: roots,
		logger: logger, wake: make(chan struct{}, 1)}
	im.ClientFor = func(ctx context.Context, code string) (downloader.Driver, error) {
		return acquire.DownloadClientFor(ctx, pool, code)
	}
	return im
}

// Wake asks for a pass now: after a grab, or a person's decision.
func (im *Importer) Wake() {
	select {
	case im.wake <- struct{}{}:
	default:
	}
}

// Run passes every minute, and when woken, until ctx ends.
func (im *Importer) Run(ctx context.Context) {
	t := time.NewTicker(interval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		case <-im.wake:
		}
		rep := im.Pass(ctx)
		if rep.Imported+rep.Refused+rep.Review+rep.Held+rep.Failed > 0 || rep.Error != "" {
			im.logger.Info("import: pass", "imported", rep.Imported, "refused", rep.Refused,
				"review", rep.Review, "held", rep.Held, "failed", rep.Failed,
				"downloading", rep.Downloading, "error", rep.Error)
		}
	}
}

// Report is one pass.
type Report struct {
	Downloading, Imported, Refused, Review, Held, Failed, Waiting int
	Error                                                         string
}

// Pass looks at every live grab once.
func (im *Importer) Pass(ctx context.Context) Report {
	im.mu.Lock()
	defer im.mu.Unlock()
	ctx, cancel := context.WithTimeout(ctx, passTimeout)
	defer cancel()
	var rep Report
	q := dbq.New(im.pool)
	if _, err := q.ReclaimStaleImports(ctx, staleMinutes); err != nil {
		rep.Error = err.Error()
		return rep
	}
	grabs, err := q.ListLiveGrabs(ctx)
	if err != nil {
		rep.Error = err.Error()
		return rep
	}
	byClient := map[string][]dbq.ListLiveGrabsRow{}
	var order []string
	for _, g := range grabs {
		if _, ok := byClient[g.Client]; !ok {
			order = append(order, g.Client)
		}
		byClient[g.Client] = append(byClient[g.Client], g)
	}
	for _, code := range order {
		im.passClient(ctx, q, code, byClient[code], &rep)
	}
	return rep
}

func (im *Importer) passClient(ctx context.Context, q *dbq.Queries, code string, gs []dbq.ListLiveGrabsRow, rep *Report) {
	wait := func(g dbq.ListLiveGrabsRow, why string) {
		rep.Waiting++
		_ = q.SetGrabProgress(ctx, dbq.SetGrabProgressParams{
			ID: g.ID, Progress: g.Progress, ImportState: StateWaiting, ImportReason: why})
	}
	d, err := im.ClientFor(ctx, code)
	if err != nil {
		for _, g := range gs {
			wait(g, fmt.Sprintf("Can't reach its download client: %v.", err))
		}
		return
	}
	hashes := make([]string, 0, len(gs))
	for _, g := range gs {
		hashes = append(hashes, g.InfoHash)
	}
	ts, err := d.Torrents(ctx, hashes...)
	if err != nil {
		for _, g := range gs {
			wait(g, fmt.Sprintf("Can't reach its download client %s: %s", code, indexer.FirstLine(err.Error())))
		}
		return
	}
	byHash := map[string]downloader.Torrent{}
	for _, t := range ts {
		byHash[indexer.NormalizeHash(t.Hash)] = t
	}
	for _, g := range gs {
		if ctx.Err() != nil {
			return
		}
		switch g.ImportState {
		case StateReview, StateFailed, StateImporting:
			continue
		}
		t, ok := byHash[indexer.NormalizeHash(g.InfoHash)]
		if !ok {
			continue // gone from the client: the lifecycle settles it (#5523)
		}
		if !g.WantID.Valid {
			wait(g, "Its album is no longer wanted.")
			continue
		}
		if !finished(ctx, d, t) {
			rep.Downloading++
			_ = q.SetGrabProgress(ctx, dbq.SetGrabProgressParams{
				ID: g.ID, Progress: float32(min(t.Progress, 0.999)), ImportState: StateDownloading})
			continue
		}
		claimed, err := q.ClaimGrabImport(ctx, g.ID)
		if err != nil {
			continue // claimed by a pass racing this one, or ended
		}
		switch out := im.importOne(ctx, q, claimed, d, t); out {
		case StateImported:
			rep.Imported++
		case StateRefused:
			rep.Refused++
		case StateReview:
			rep.Review++
		case StateHeld:
			rep.Held++
		case StateFailed:
			rep.Failed++
		case StateWaiting:
			rep.Waiting++
		}
	}
}

// upStates are qBittorrent's states for a torrent that has every byte and
// is seeding, paused or queued to seed. Exchange's states arrive in the same
// vocabulary.
var upStates = map[string]bool{
	"uploading": true, "stalledUP": true, "pausedUP": true, "stoppedUP": true,
	"queuedUP": true, "forcedUP": true, "checkingUP": true,
}

// finished: whole, in a seeding state, and no wanted file short of bytes.
func finished(ctx context.Context, d downloader.Driver, t downloader.Torrent) bool {
	if !t.Done() || !upStates[t.State] {
		return false
	}
	if fp, ok := d.(downloader.FileProgress); ok {
		if states, err := fp.FileStates(ctx, t.Hash); err == nil {
			for _, f := range states {
				if f.Unfinished() {
					return false
				}
			}
		}
	}
	return true
}

// Detail is what grabs.import_detail keeps: the verdict and what it rested
// on, the edition chosen, and what happened to the files.
type Detail struct {
	Verdict     Verdict  `json:"verdict"`
	Edition     string   `json:"edition,omitempty"`
	EditionMBID string   `json:"edition_mbid,omitempty"`
	Files       []string `json:"files,omitempty"`
	Placed      int      `json:"placed,omitempty"`
	Skipped     int      `json:"skipped,omitempty"`
	Untagged    int      `json:"untagged,omitempty"`
	// RehomedFrom: the album it was grabbed for, when it turned out to be
	// another wanted album.
	RehomedFrom string `json:"rehomed_from,omitempty"`
}

// importOne checks and files one finished download and records the outcome.
func (im *Importer) importOne(ctx context.Context, q *dbq.Queries, g dbq.Grab, d downloader.Driver, t downloader.Torrent) string {
	record := func(state, reason, end string, det Detail, files int, dest string) string {
		body, _ := json.Marshal(det)
		// The outcome is written even when the pass's context is ending:
		// a claimed grab left "importing" waits half an hour to be retried.
		wctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 15*time.Second)
		defer cancel()
		if _, err := q.RecordImport(wctx, dbq.RecordImportParams{
			ID: g.ID, ImportState: state, ImportReason: reason, ImportDetail: body,
			ImportedFiles: int32(files), Destination: dest, EndReason: end,
		}); err != nil {
			im.logger.Warn("import: record outcome", "grab", g.Name, "state", state, "err", err)
		}
		im.logger.Info("import: "+state, "grab", g.Name, "reason", reason)
		return state
	}

	w, err := im.want(ctx, q, g.WantID)
	if err != nil {
		return record(StateWaiting, "Its album could not be read: "+err.Error(), "", Detail{}, 0, "")
	}
	editions, err := im.editions(ctx, w.GroupMBID)
	if err != nil {
		return record(StateWaiting, "MusicBrainz could not be reached to check it against the album; trying again shortly.",
			"", Detail{}, 0, "")
	}
	src := t.ContentPath
	if src == "" {
		src = filepath.Join(g.SavePath, t.Name)
	}
	files, printed, err := inspect(ctx, src, im.Printer)
	if err != nil {
		return record(StateFailed, fmt.Sprintf("Couldn't read the download at %s: %v", src, err), "", Detail{}, 0, "")
	}
	v := Verify(w, editions, files, printed, g.ImportForced)
	det := Detail{Verdict: v}
	for _, f := range files {
		det.Files = append(det.Files, f.Rel)
	}
	if v.Edition >= 0 {
		det.Edition, det.EditionMBID = describe(editions[v.Edition]), editions[v.Edition].MBID
	}

	switch v.Outcome {
	case Review:
		return record(StateReview, v.Reason, "", det, 0, "")
	case Refuse:
		if v.Code == CodeWrongAlbum {
			if out, ok := im.rehome(ctx, q, g, d, w, files, printed, record); ok {
				return out
			}
		}
		im.refuse(ctx, q, g, d, w, v.Reason, pgtype.UUID{})
		return record(StateRefused, v.Reason, StateRefused, det, 0, "")
	}
	return im.place(ctx, q, g, d, w, editions, files, v, det, record)
}

type recordFunc func(state, reason, end string, det Detail, files int, dest string) string

// place files an accepted download and records it imported, or held when
// the library already had every track.
func (im *Importer) place(ctx context.Context, q *dbq.Queries, g dbq.Grab, d downloader.Driver, w Want,
	editions []Edition, files []File, v Verdict, det Detail, record recordFunc) string {
	root, err := im.root(ctx, q)
	if err != nil {
		return record(StateFailed, "Not imported: "+err.Error()+".", "", det, 0, "")
	}
	var e *Edition
	if v.Edition >= 0 {
		e = &editions[v.Edition]
	}
	plan := Plan(root, w, e, files, v.Pairs)
	if w.Upgrade != "" {
		return im.placeUpgrade(ctx, q, g, d, w, e, files, plan, v, det, record)
	}
	heldRec, heldPos, err := im.held(ctx, q, w, plan)
	if err != nil {
		return record(StateWaiting, "Couldn't read what the library holds: "+err.Error(), "", det, 0, "")
	}

	var placed []string
	for _, pl := range plan {
		if (pl.Recording != "" && heldRec[pl.Recording]) || heldPos[[2]int{pl.Disc, pl.Track}] {
			det.Skipped++
			continue
		}
		tags := Tags{Title: pl.Title, Artist: w.Artist, Album: w.Title, AlbumArtist: w.Artist, Year: w.Year,
			Disc: pl.Disc, Discs: pl.Discs, Track: pl.Track, Tracks: pl.Tracks,
			GroupMBID: w.GroupMBID, RecordingMBID: pl.Recording, ArtistMBID: w.ArtistMBID}
		if e != nil {
			tags.ReleaseMBID = e.MBID
		}
		tagged, err := Place(ctx, files[pl.File].Path, pl.Dest, tags)
		if errors.Is(err, ErrOccupied) {
			det.Skipped++
			continue
		}
		if err != nil {
			for _, p := range placed {
				_ = os.Remove(p)
			}
			return record(StateFailed, fmt.Sprintf("Copying into the library failed, and nothing was left behind: %v", err),
				"", det, 0, "")
		}
		placed = append(placed, pl.Dest)
		if !tagged {
			det.Untagged++
		}
	}
	det.Placed = len(placed)
	if len(placed) == 0 {
		im.removeFromClient(ctx, d, g)
		return record(StateHeld, fmt.Sprintf("The library already holds every track of %s, so nothing was imported.", w.Title),
			StateHeld, det, 0, "")
	}

	if _, err := im.Scanner.ScanFiles(ctx, placed); err != nil {
		// The watcher and the next scan find the files anyway.
		im.logger.Warn("import: scan the new files", "grab", g.Name, "err", err)
	}
	if err := q.MarkWantSearched(ctx, g.WantID); err != nil {
		im.logger.Warn("import: mark the want searched", "grab", g.Name, "err", err)
	}
	dest := filepath.Dir(placed[0])
	if e != nil && e.Media() > 1 {
		dest = filepath.Dir(dest)
	}
	reason := fmt.Sprintf("Imported %s into %s. %s", tracks(len(placed)), dest, v.Reason)
	if det.Skipped > 0 {
		reason += fmt.Sprintf(" %s already in the library left as they were.", tracks(det.Skipped))
	}
	if det.Untagged > 0 {
		reason += fmt.Sprintf(" %s copied with their own tags: ffmpeg could not rewrite them.", tracks(det.Untagged))
	}
	return record(StateImported, reason, StateImported, det, len(placed), dest)
}

// rehome imports a download that turned out to be another wanted album,
// named by its tags, for that album, when its own grab is not running.
func (im *Importer) rehome(ctx context.Context, q *dbq.Queries, g dbq.Grab, d downloader.Driver,
	from Want, files []File, printed bool, record recordFunc) (string, bool) {
	group := majority(files, func(f File) string { return f.GroupMBID })
	if group == "" {
		return "", false
	}
	others, err := q.OpenWantsForGroups(ctx, []string{group})
	if err != nil || len(others) == 0 {
		return "", false
	}
	other := others[0]
	w, err := im.want(ctx, q, other.ID)
	if err != nil {
		return "", false
	}
	editions, err := im.editions(ctx, w.GroupMBID)
	if err != nil {
		return "", false
	}
	v := Verify(w, editions, files, printed, false)
	if v.Outcome != Accept {
		return "", false
	}
	moved, err := q.RehomeGrab(ctx, dbq.RehomeGrabParams{ID: g.ID, WantID: other.ID})
	if err != nil {
		return "", false
	}
	if err := q.SearchWantAgain(ctx, g.WantID); err != nil {
		im.logger.Warn("import: search the first album again", "want", from.Title, "err", err)
	}
	det := Detail{Verdict: v, RehomedFrom: from.Title}
	for _, f := range files {
		det.Files = append(det.Files, f.Rel)
	}
	if v.Edition >= 0 {
		det.Edition, det.EditionMBID = describe(editions[v.Edition]), editions[v.Edition].MBID
	}
	v.Reason = fmt.Sprintf("Grabbed for %s, it is %s, which is wanted too. %s", from.Title, w.Title, v.Reason)
	det.Verdict = v
	return im.place(ctx, q, moved, d, w, editions, files, v, det, record), true
}

// refuse blocks the release, asks for the album to be searched again and
// removes the download with its files: it is not the album.
func (im *Importer) refuse(ctx context.Context, q *dbq.Queries, g dbq.Grab, d downloader.Driver, w Want, why string, by pgtype.UUID) {
	if err := acquire.NewMemory(im.pool).Block(ctx, dbq.BlockReleaseParams{
		InfoHash: g.InfoHash, Name: g.Name, Reason: acquire.BlockImport, Detail: why, WantID: g.WantID, BlockedBy: by,
	}); err != nil {
		im.logger.Warn("import: block the release", "grab", g.Name, "err", err)
	}
	if err := q.SearchWantAgain(ctx, g.WantID); err != nil {
		im.logger.Warn("import: search again", "want", w.Title, "err", err)
	}
	im.removeFromClient(ctx, d, g)
}

func (im *Importer) removeFromClient(ctx context.Context, d downloader.Driver, g dbq.Grab) {
	if d == nil || g.InfoHash == "" {
		return
	}
	if err := d.Delete(ctx, true, g.InfoHash); err != nil {
		im.logger.Warn("import: remove the download", "grab", g.Name, "err", err)
	}
}

// Refuse is a person's "refuse" on a download the check held for them, or
// one that failed: it is refused as the check would have.
func (im *Importer) Refuse(ctx context.Context, grabID, by pgtype.UUID) (dbq.Grab, error) {
	q := dbq.New(im.pool)
	g, err := q.GetGrab(ctx, grabID)
	if err != nil {
		return dbq.Grab{}, err
	}
	if g.EndedAt.Valid || (g.ImportState != StateReview && g.ImportState != StateFailed) {
		return dbq.Grab{}, ErrNotDecidable
	}
	var w Want
	if g.WantID.Valid {
		w, _ = im.want(ctx, q, g.WantID)
	}
	d, err := im.ClientFor(ctx, g.Client)
	if err != nil {
		d = nil
	}
	why := "A person refused it: it is not the album."
	im.refuse(ctx, q, g, d, w, why, by)
	out, err := q.RecordImport(ctx, dbq.RecordImportParams{
		ID: g.ID, ImportState: StateRefused, ImportReason: why, ImportDetail: g.ImportDetail,
		EndReason: StateRefused,
	})
	im.Wake()
	return out, err
}

// ErrNotDecidable: the grab is not waiting on a person.
var ErrNotDecidable = errors.New("importer: that download is not waiting on a decision")

func (im *Importer) want(ctx context.Context, q *dbq.Queries, id pgtype.UUID) (Want, error) {
	row, err := q.GetWantForSearch(ctx, id)
	if err != nil {
		return Want{}, err
	}
	w := Want{
		GroupMBID: row.ReleaseGroupMbid, ArtistMBID: row.ArtistMbid, Artist: row.ArtistName, Title: row.Title,
		Year: yearOf(row.FirstReleaseDate), PrimaryType: row.PrimaryType, SecondaryTypes: row.SecondaryTypes,
	}
	if row.FulfilledAt.Valid && row.Upgrade != "" && row.AlbumID.Valid {
		w.Upgrade, w.AlbumID = row.Upgrade, row.AlbumID
	}
	return w, nil
}

func (im *Importer) editions(ctx context.Context, group string) ([]Edition, error) {
	if im.Catalogue == nil {
		return nil, errors.New("no catalogue")
	}
	eds, err := im.Catalogue.Releases(ctx, group)
	if err != nil {
		return nil, err
	}
	out := make([]Edition, 0, len(eds.Editions))
	for _, e := range eds.Editions {
		ed := Edition{MBID: e.Release.Mbid, Title: e.Release.Title, Status: e.Release.Status,
			Date: e.Release.ReleaseDate, Country: e.Release.Country}
		for _, t := range e.Tracks {
			ed.Tracks = append(ed.Tracks, Track{Medium: int(t.MediumPosition), Format: t.MediumFormat,
				Position: int(t.Position), Title: t.Title, RecordingMBID: t.RecordingMbid, LengthMs: int(t.LengthMs)})
		}
		slices.SortStableFunc(ed.Tracks, func(a, b Track) int {
			if a.Medium != b.Medium {
				return a.Medium - b.Medium
			}
			return a.Position - b.Position
		})
		out = append(out, ed)
	}
	return out, nil
}

// root is the library folder new music is filed into.
func (im *Importer) root(ctx context.Context, q *dbq.Queries) (string, error) {
	if len(im.Roots) == 0 {
		return "", errors.New("there is no library folder to file new music into (MINSTREL_LIBRARY_SCAN_PATHS)")
	}
	s, err := q.GetAcquisitionSettings(ctx)
	if err == nil && s.ImportRoot != "" && slices.Contains(im.Roots, s.ImportRoot) {
		return s.ImportRoot, nil
	}
	return im.Roots[0], nil
}

// held reads which of the planned tracks the library already has: by
// recording, and by disc and track of an album of the same group.
func (im *Importer) held(ctx context.Context, q *dbq.Queries, w Want, plan []Placement) (map[string]bool, map[[2]int]bool, error) {
	var recs []string
	for _, pl := range plan {
		if pl.Recording != "" {
			recs = append(recs, pl.Recording)
		}
	}
	rec := map[string]bool{}
	if len(recs) > 0 {
		got, err := q.HeldRecordings(ctx, recs)
		if err != nil {
			return nil, nil, err
		}
		for _, r := range got {
			rec[r] = true
		}
	}
	pos := map[[2]int]bool{}
	got, err := q.HeldPositions(ctx, &w.GroupMBID)
	if err != nil {
		return nil, nil, err
	}
	for _, p := range got {
		pos[[2]int{int(p.Disc), int(p.Track)}] = true
	}
	return rec, pos, nil
}

func yearOf(date string) int {
	if len(date) < 4 {
		return 0
	}
	y, err := strconv.Atoi(date[:4])
	if err != nil {
		return 0
	}
	return y
}

func tracks(n int) string {
	if n == 1 {
		return "1 track"
	}
	return strconv.Itoa(n) + " tracks"
}